Archive

Posts Tagged ‘simultaneously’

Concurrency, when several threads fight for the access to a resource [ example in C ]

January 30, 2014 No comments

If we’re creating a multi-thread application and we’re also sharing information between the main thread and the secondary thread, or between threads, you must have in mind the type of access to that information.
For example, if we will only allow one thread to write on a variable and the other will just read we won’t have any problem in most cases, but if any thread can write a value at any time, we must be careful. If some threads are willing to write a variable at almost the same time, only the last value written will remain.

Another example, we have a film collection software, and at the moment we have 50 films stored. Another thread is going to synchronize an Internet server, but while the synchronization is running, we add 3 films more. The thread synchronizing may see 50 films, but the films sent can be a mix of the old and new films, so the server will think we’ve removed some films and we will have a problem.
In this case, we must protect the access to the critic section (our film list), so when we are adding data, the other thread can’t sync, and when the other thread is syncing, we must wait before adding anything. We will use for that mutual exclusion or mutex.

To try to make a visible example, we’re incrementing numbers, but we will insert a CPU eater task between the read of the value and the writing of the new value. The CPU-eater task can finish in a variable time interval, so one threads will finish this task before others. The desired result is the number incrementing to 10, but the real one differs a bit:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
#include <string.h>

struct thread_vars_t
{
  int number;
};

int numberExists(int arr[], int current)
{
  int i;

  for (i=0; i<current; ++i)
    {
      if (arr[current]==arr[i])
    return 1;
    }

  return 0;
}

void numberSearch()
{
  /* A time taking task */
  int numbers[100];
  int i;

  for (i=0; i<100; ++i)
    {
      numbers[i] = rand()%101;
      while (numberExists(numbers, i))
    numbers[i] = rand()%101;
    }
}


void *newtask(void *_number)
{
  struct thread_vars_t *vars = _number;

  int number = vars->number;
  numberSearch();
  vars->number = number+1;

  printf ("THREAD: number = %d\n", vars->number);

  pthread_exit(NULL);
}

int main (int argc, char *argv[])
{
   pthread_t thread;
   int rc;
   int i;
   struct thread_vars_t *vars = malloc (sizeof(struct thread_vars_t));

   vars->number = 0;

   printf ("Main process just started.\n");
   for (i=0; i<10; ++i)
     {
       rc = pthread_create(&thread, NULL, newtask, vars);
       if (rc)
     {
       printf("ERROR in pthread_create(): %d\n", rc);
       exit(-1);
     }
     }

   printf ("Main process about to finish.\n");
   /* Last thing that main() should do */
   pthread_exit(NULL);
}

Your result can be more or less like that:

$ ./sharedvar
Main process just started.
THREAD : number = 1
THREAD : number = 1
THREAD : number = 2
THREAD : number = 2
THREAD : number = 3
Main process about to finish.
THREAD : number = 4
THREAD : number = 4
THREAD : number = 3
THREAD : number = 4
THREAD : number = 4

What has happened? Some threads read the variable when it was 0 (two occasions), so they both incremented to 1, others read the value when it was 1 (another two ones), and incremented to 2, in other cases, the variable was 2 and was incremented to 3…
So, several threads read the same value and when writing the new value, we didn’t have in mind the value could have changed by another thread while we were working. That is the race condition.

How can we fix that? The solution is coding structures that block access to the resource when it’s being used. For example, it some other thread has read the value of the variable, no other can, until a new value is written.
Do we lose performance? Yes, a bit, because we are waiting for other tasks instead of working together. But we avoid undesirable situations like the example before. But the threads may do also some other things outside the critical section, and this work can be done simultaneously. We will only block the critical section (when working on a number), when we will block other threads with a mutex.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
#include <string.h>

struct thread_vars_t
{
  int number;
  pthread_mutex_t mutex;
};

int numberExists(int arr[], int current)
{
  int i;

  for (i=0; i<current; ++i)
    {
      if (arr[current]==arr[i])
    return 1;
    }

  return 0;
}

void numberSearch()
{
  /* A time taking task */
  int numbers[100];
  int i;

  for (i=0; i<100; ++i)
    {
      numbers[i] = rand()%101;
      while (numberExists(numbers, i))
    numbers[i] = rand()%101;
    }
}


void *newtask(void *_number)
{
  struct thread_vars_t *vars = _number;

  /* BLOCK */
  pthread_mutex_lock(&vars->mutex);
  /* BLOCK */

  int number = vars->number;
  numberSearch();
  vars->number = number+1;

  /* UNBLOCK */
  pthread_mutex_unlock(&vars->mutex);
  /* UNBLOCK */

  printf ("THREAD: number = %d\n", vars->number);

  pthread_exit(NULL);
}

int main (int argc, char *argv[])
{
   pthread_t thread;
   int rc;
   int i;
   struct thread_vars_t *vars = malloc (sizeof(struct thread_vars_t));

   pthread_mutex_init(&vars->mutex, NULL);
   vars->number = 0;

   printf ("Main process just started.\n");
   for (i=0; i<10; ++i)
     {
       rc = pthread_create(&thread, NULL, newtask, vars);
       if (rc)
     {
       printf("ERROR in pthread_create(): %d\n", rc);
       exit(-1);
     }
     }

   printf ("Main process about to finish.\n");
   /* Last thing that main() should do */
   pthread_exit(NULL);
}

And the result will be like this:

$ ./simplemutex
Main process just started.
THREAD: number = 1
Main process about to finish.
THREAD: number = 2
THREAD: number = 3
THREAD: number = 4
THREAD: number = 5
THREAD: number = 6
THREAD: number = 7
THREAD: number = 8
THREAD: number = 9
THREAD: number = 10

Photo: Daryl L. Hunter (Flickr) CC-by

Non-blocking TCP client in C, where we can send and receive data at every moment

January 10, 2014 No comments

If you are creating a program which interacts with a server, this may interest you. First, we’re going to make a simple TCP client, we connect to a server and it will allow us to send whatever we write on our keyboard. We can’t receive anything from the server right now, it’s just a first step to reach our goal:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <netdb.h>
#include <arpa/inet.h>
#include <errno.h>
#include <unistd.h>

#define MENS_MAX_LEN 500

void usage()
{
  fprintf (stderr, "Wrong arguments. Must give two:\n");
  fprintf (stderr, "tcpclient SERVER PORT\n\n");
  fprintf (stderr, "For example:\n");
  fprintf (stderr, "  tcpclient totaki.com 80\n");
  exit(1);
}

void panic(char *msg)
{
  fprintf (stderr, "Fatal error: %s (errno %d, %s)\n", msg, errno, strerror(errno));
  exit(2);
}

int main(int argc, char *argv[])
{
  char *server;
  int port;
  int socketfd;
  int finish = 0;
  struct sockaddr_in serverAddress, clientAddress;
  struct hostent *h;
  char mens [MENS_MAX_LEN];
  size_t recvsize;

  if (argc<3)
    usage();

  server = argv[1];
  port = atoi(argv[2]);

  /* Create a TCP socket */
  socketfd = socket(AF_INET, SOCK_STREAM, 0);
  if (socketfd==-1)
    panic("Failed to create socket");

  /* Sets client address */
  clientAddress.sin_family = AF_INET;
  clientAddress.sin_addr.s_addr = htonl(INADDR_ANY);
  clientAddress.sin_port = htons(0);

  /* Bind the address to the socket */
  if (bind(socketfd, (struct sockaddr *)&clientAddress, sizeof(clientAddress))==-1)
    panic("Binding address");

  /* Get the hostname address */
  h = gethostbyname(server);

  /* Sets server address */
  serverAddress.sin_family = h->h_addrtype;
  memcpy((char*) &serverAddress.sin_addr.s_addr, h->h_addr_list[0], h->h_length);
  serverAddress.sin_port = htons(port);

  if (connect(socketfd, (struct sockaddr * )&serverAddress, sizeof(serverAddress))==-1)
    panic("Cannot connect");

  do
    {
      fgets(mens, MENS_MAX_LEN, stdin);

      if (send(socketfd, mens, strlen(mens)+1, 0)==-1)
    panic ("Cannot sent message");

    } while (!finish);      /* Never finish */

  return EXIT_SUCCESS;
}

We’ll analyze the main() function directly, because usage() just display text and panic() makes the program exit on an error.
The first thing we do is to determine the server and port we will connect to, they both will be the first and second arguments to our program. Then we create the socket and build the client address. The port is 0 to let the system pick any port, we don’t mind which port is picked. Then bind() will assign the address to the socket.

Then, create the server address. We will use getbyhostname() before because we may resolve the server address first (if it comes with a hostname, or even if it comes as a IP address we will have to translate this address to a suitable format), so let’s connect() and begin sending messages.

In this case we’re using fgets() to ask the user for information through the standard input (usually keyboard) and then send() it, we must set the size to strlen(mens)+1 to include the string terminator ‘\0′ at the end of the string being sent.

Note: fgets() will send the line feed at the end when we press return. We can use a trim() function or filter it with:

1
2
if (mens[strlen(mens)-1]=='\n')
   mens[strlen(mens)-1] = '\0';

And to receive a message from the server, we must:

1
2
3
4
5
size_t recvsize;
recvsize = recv(socketfd, mens, MENS_MAX_LEN, 0);
if (recvsize==-1)
  panic("Cannot receive information");
mens[recvsize] = '\0';

Reception is similar to sending, but this time we don’t know the size of the message, so we provide a maximum size (or the size of our buffer), if the size of the data we are receiving is greater than our buffer, we can continue the message calling again and again recv().
As this message may not have a terminator, to display it properly on screen with printf() we must put the terminator manually at the end of the string.

It’s easy. The biggest problem, comes when we want the application to send and receive information whenever there is data on each side (sending or reception), making our application full duplex. With this code, right now, we just can give turns, make the user send information sometimes, and sometimes the application will receive data.
We will make a more complex example using select(), this function will detect whenever there is data waiting on some descriptors, and we can later choose whether to read the socket or the keyboard. But the most interesting is the timeout, we can give select() a wait timeout, when the function will stop waiting and the program can continue, it won’t block indefinitely waiting for data.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <sys/types.h>
#include <sys/select.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <netdb.h>
#include <arpa/inet.h>
#include <errno.h>
#include <unistd.h>

#define MENS_MAX_LEN 500

void usage()
{
  fprintf (stderr, "Wrong arguments. Must give two:\n");
  fprintf (stderr, "tcpclient SERVER PORT\n\n");
  fprintf (stderr, "For example:\n");
  fprintf (stderr, "  tcpclient totaki.com 80\n");
  exit(1);
}

void panic(char *msg)
{
  fprintf (stderr, "Fatal error: %s (errno %d, %s)\n", msg, errno, strerror(errno));
  exit(2);
}

int main(int argc, char *argv[])
{
  char *server;
  int port;
  int socketfd;
  int finish = 0;
  struct sockaddr_in serverAddress, clientAddress;
  struct hostent *h;
  char mens [MENS_MAX_LEN];
  fd_set readmask;
  struct timeval timeout;
  size_t recvsize;

  if (argc<3)
    usage();

  server = argv[1];
  port = atoi(argv[2]);

  /* Create a TCP socket */
  socketfd = socket(AF_INET, SOCK_STREAM, 0);
  if (socketfd==-1)
    panic("Failed to create socket");

  /* Sets client address */
  clientAddress.sin_family = AF_INET;
  clientAddress.sin_addr.s_addr = htonl(INADDR_ANY);
  clientAddress.sin_port = htons(0);

  /* Bind the address to the socket */
  if (bind(socketfd, (struct sockaddr *)&clientAddress, sizeof(clientAddress))==-1)
    panic("Binding address");

  /* Get the hostname address */
  h = gethostbyname(server);

  /* Sets server address */
  serverAddress.sin_family = h->h_addrtype;
  memcpy((char*) &serverAddress.sin_addr.s_addr, h->h_addr_list[0], h->h_length);
  serverAddress.sin_port = htons(port);

  if (connect(socketfd, (struct sockaddr * )&serverAddress, sizeof(serverAddress))==-1)
    panic("Cannot connect");

  do
    {
      /* We must set all this information on each select we do */
      FD_ZERO(&readmask);   /* empty readmask */
      /* Then we put all the descriptors we want to wait for in a */
      /* mask = readmask */
      FD_SET(socketfd, &readmask);
      FD_SET(STDIN_FILENO, &readmask); /* STDIN_FILENO = 0 (standard input) */
      /* Timeout, we will stop waiting for information */
      timeout.tv_sec=0;
      timeout.tv_usec=100000;

      /* The first parameter is the biggest descriptor+1. The first one
       was 0, so every other descriptor will be bigger.*/

      /* readfds = &readmask */
      /* writefds = we are not waiting for writefds */
      /* exceptfds = we are not waiting for exception fds */
      if (select(socketfd+1, &readmask, NULL, NULL, &timeout)==-1)
    panic("Error on SELECT");

      /* If something was received */
      if (FD_ISSET(socketfd, &readmask))
    {
      recvsize = recv(socketfd, mens, MENS_MAX_LEN, 0);
      if (recvsize==-1)
        panic("Cannot receive information");
      mens[recvsize] = '\0';
      printf (">> %s\n", mens);
    }

      /* If something was written by the user */
      if (FD_ISSET(STDIN_FILENO, &readmask))
    {
      fgets(mens, MENS_MAX_LEN, stdin);

      if (send(socketfd, mens, strlen(mens)+1, 0)==-1)
        panic ("Cannot sent message");
    }
    } while (!finish);

  return EXIT_SUCCESS;
}

Photo: Julien Gong Min (Flickr) CC-by

Top