This is part 2 of a series of articles on socket programming on UNIX. In part 1 you have read about basic techniques for writing simple TCP-based clients and servers. You know the calling sequence of system calls for servers and clients and know how to serve multiple clients by forking new processes for each connection request. Today I want to show you what it takes to write a multi-threaded server facilitating POSIX Threads (pthreads) and how we use the recv and send system calls instead of read and write.
A brief introduction to POSIX Threads
A thread is a stream of instructions that can be scheduled as an independent unit. It is easier to understand a thread within the context of a process. A process is created by an operating system; a process contains information about resources, such as process id, file descriptors, etc.; in addition it contains information pertaining to the execution state, such as program counter and stack. The concept of a thread requires that we make a separation between these two kinds of information in a process:
- resources – these are available to the entire process e.g., program instructions, global data, working directory, etc.
- schedulable entities – these would include program counters and stacks. A thread is an entity within a process which consists of the schedulable part of the process.
Each process can have many threads which share the resources within a process (address space, for example). As compared to the cost of creating a process, a thread can be created with much less expense, because there is much less intervention on the part of the operating system.
What are pthreads?
An unambiguously defined interface is essential if threads are to be useful to programmers that wish to take advantage of the capabilities provided by threads. While hardware vendors each have their own implementation of threads which differ from one another, the emerging standard on Unix systems is specified by IEEE POSIX 1003.1c -1995. A threads implementation that follows this standard is referred to as pthreads. Most hardware vendors now tend to offer pthreads. in addition to their proprietary API’s. There are several drafts of the above standard, and most modern UNIX systems and its derivatives such as AIX 4.1+, Solaris 7, Tru64 Unix 5.1+, etc. follows draft 7 and above of the POSIX standard. It is important to be aware of the draft number of a given implementation, because there are differences between drafts which can cause problems.
The pthreads API
The API is defined in the ANSI/IEEE POSIX 1003.1 – 1995 standard. As compared to MPI, this standard is not freely available on the Web, and must be purchased from IEEE. An immediate consequence is that a full listing of the pthread functions is only available from the Single UNIX Specification.
Creating and Terminating Threads
Four function calls are involved in managing a thread’s lifetime:
#include <pthread.h> int pthread_create(pthread_t *thread, const pthread_attr_t *attr, void *(*start_routine)(void*), void *arg); void pthread_exit(void *value_ptr); int pthread_attr_init(pthread_attr_t *attr); int pthread_attr_destroy(pthread_attr_t *attr);
Initially, your main() program comprises a single, default thread. All other threads must be explicitly created. pthread_create creates a new thread and makes it executable. Typically, threads are first created from within main() inside a single process. Once created, threads are peers, and may create other threads.
pthread_create() takes the following arguments:
thread– An opaque, unique identifier for the new thread returned by the subroutine.attr– An opaque attribute object that may be used to set thread attributes. You can specify a thread attributes object, or NULL for the default values.start_routine– The C routine that the thread will execute once it is created.arg– A single argument that may be passed tostart_routine. It must be passed by reference as a pointer cast of type void.NULLmay be used if no argument is to be passed.
Thread Attributes
By default, a thread is created with certain attributes. Some of these attributes can be changed by the programmer via the thread attribute object. pthread_attr_init and pthread_attr_destroy are used to initialize/destroy the thread attribute object. Other routines are then used to query/set specific attributes in the thread attribute object.
Terminating Threads
There are several ways in which a pthread may be terminated:
- The thread returns from its starting routine (the main routine for the initial thread)
- The thread makes a call to the
pthread_exitsubroutine (covered below) - The thread is canceled by another thread via the
pthread_cancelroutine (not covered here) - The entire process is terminated due to a call to either the
execorexitsubroutines
pthread_exit is used to explicitly exit a thread. Typically, the pthread_exit() routine is called after a thread has completed its work and is no longer required to exist. If main() finishes before the threads it has created, and exits with pthread_exit(), the other threads will continue to execute. Otherwise, they will be automatically terminated when main() finishes. You may optionally specify a termination status, which is stored as a void pointer for any thread that may join the calling thread.
Please note that the pthread_exit() routine does not close files; any files opened inside the thread will remain open after the thread is terminated.
Example: Creating and terminating threads
This simple example code creates 10 threads with the pthread_create() routine. Each thread prints a Hello World! message, and then terminates with a call to pthread_exit().
#include <pthread.h> #include <stdlib.h> #define NUM_THREADS 10 void *print_hello(void *tid) { printf("nThis is thread %d: Hello World!n", tid); pthread_exit(NULL); } int main (int argc, char **argv) { pthread_t threads[NUM_THREADS]; int rc, i; for (i = 0; i < NUM_THREADS; i++){ printf("Creating thread %dn", i); rc = pthread_create(&threads[i], NULL, print_hello, (void *)i); if (rc) { printf("ERROR; pthread_create() returned: %dn", rc); exit(-1); } } pthread_exit(NULL); }
Argument passing with threads
As you may have noticed, we can specify a callback function which is called when the associated thread is created. How do we pass an argument to the callback function? The pthread_create() routine permits us to pass one argument to the thread start routine. For cases where multiple arguments must be passed, this limitation is easily overcome by creating a structure which contains all of the arguments, and then passing a pointer to that structure in the routine. All arguments must be passed by reference and cast to (void *).
#include <pthread.h> #include <stdlib.h> #include <unistd.h> #define NUM_THREADS 3 char *messages[NUM_THREADS]; struct data { int tid; int sum; char *message; }; struct data data_array[NUM_THREADS]; void *PrintHello(void *arg) { int taskid, sum; char *hello_msg; struct data *my_data; sleep(1); my_data = (struct data *)arg; taskid = my_data->tid; sum = my_data->sum; hello_msg = my_data->message; printf("Thread %d: %s Sum=%dn", taskid, hello_msg, sum); pthread_exit(NULL); } int main(int argc, char **argv) { pthread_t threads[NUM_THREADS]; int *taskids[NUM_THREADS]; int rc, t, sum; sum = 0; messages[0] = "English: Hello World!"; messages[1] = "French: Bonjour, le monde!"; messages[2] = "German: Guten Tag, Welt!"; for (t = 0; t < NUM_THREADS; t++) { sum = sum + t; data_array[t].tid = t; data_array[t].sum = sum; data_array[t].message = messages[t]; printf("Creating thread %dn", t); rc = pthread_create(&threads[t], NULL, PrintHello, (void *) &data_array[t]); if (rc) { printf("ERROR; pthread_create() returned is %dn", rc); exit(-1); } } pthread_exit(NULL); }
It’s quite easy to pass data to callback functions of threads. Just provide the data as a void pointer and cast it back in the function. That’s it.
Joining Threads
Joining a thread is a simple way to notify the worker thread that a thread which has been created on behalf of the worker has finished. To join a thread we call phread_join() passing the thread ID of the thread to wait for as its argument:
#include <pthread.h> int pthread_join(pthread_t thread, void **value_ptr);
thread identifies the thread we want to wait for and value_ptr stores the status of the thread that is returned by pthread_exit()
The main thread creates other threads and is able to wait for them to finish. When controlling thread calls pthread_join() it is notified when a particular thread is destroyed. The interaction between both threads is outlined in the figure below:
Joining Threads
Example: Joining threads
#include <pthread.h> #include <stdlib.h> #include <unistd.h> #define NUM_THREADS 3 void *worker_func(void *null) { int i; double result = 0.0; for (i = 0; i < 1000000; i++) result = result + (double)random(); printf("Thread result = %en", result); pthread_exit((void *) 0); } int main(int argc, char **argv) { pthread_t thread[NUM_THREADS]; pthread_attr_t attr; int rc, t, status; pthread_attr_init(&attr); /* threads are joinable by default; demo only */ pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE); for (t = 0; t < NUM_THREADS; t++) { printf("Creating thread %dn", t); rc = pthread_create(&thread[t], &attr, worker_func, NULL); if (rc) { printf("ERROR; pthread_create() returned %dn", rc); exit(-1); } } /* free attribute and wait for the other threads */ pthread_attr_destroy(&attr); for(t = 0; t < NUM_THREADS; t++) { rc = pthread_join(thread[t], (void **)&status); if (rc) { printf("ERROR pthread_join() returned: %dn", rc); exit(-1); } printf("Join OK with thread %d status = %dn", t, status); } pthread_exit(NULL); }
Although not explicitly mentioned in the text we use pthread_attr_setdetachstate to ensure that the thread is joinable. POSIX defines threads to be joinable by default, so this has only demonstrative purposes. The detach state of a thread specifies whether the controlling thread is able to watch its execution or if it is completely depended of its controller. In this case a thread is detached. The state, among other attributes, is stored in a variable of the opaque data type pthread_attr_t which must be initialized by calling pthread_attr_init before used first.
As you can see threads deserve a complete article on its own. They can increase a program’s complexity significantly, so a fundamental understanding of threads is essential if used in larger applications.
Multi-threaded server
Let’s get back to our sockets. To write a multi-threaded server we will need at least two of the functions described in the last section: pthread_create() and pthread_join(). The server is structured as follows: main() calls client_wait setting up the address structures, binding the socket and putting the server in listening mode. Listening mode is actually done in function client_thread calling accept. The newly returned socket descriptor is passed the thread to serve the client. Just as we have seen in the forking server, fetching a connection from the queue must be performed in its own thread. Here is how it works conceptually:
int main(void) { socket(); client_wait(); /* setup sockaddr_in */ bind(); listen(); client_thread(); accept(); pthread_create(client_cb); pthread_join(); client_cb() /* serve client */
What follows is the code needed to write a multithreaded server. It already includes the recv and send system calls I am going to explain right after the code. So, here we go.
#include <socket.h> #include "header.h" void client_wait(const int); /* listen for requests */ void client_thread(int); /* create client thread */ void *client_cb(void *); /* client_thread callback */ int main(int argc, char* argv[]) { int sd; if ((sd = socket(AF_INET, SOCK_STREAM, 0)) == -1) err_fatal(errno); client_wait(sd); /* never returns */ return 0; } void client_wait(const int sd) { struct sockaddr_in sa; memset((char *)&sa, 0, sizeof(sa)); sa.sin_family = AF_INET; sa.sin_port = htons(PORT); sa.sin_addr.s_addr = INADDR_ANY; if (bind(sd, (struct sockaddr *)&sa, sizeof(sa)) == -1) err_fatal(errno); if ((listen(sd, 20)) == -1) err_fatal(errno); printf("Listening...n"); client_thread(sd); } void client_thread(int sd) { struct sockaddr_in sa; int ad;, th_addr_size; pthread_t client_th; while (1) { if ((ad = accept(sd, (struct sockaddr *)&sa, &th_addr_size)) == -1) rr_fatal(errno); if (pthread_create(client_th, NULL, client_cb, &ad)) err_fatal(errno); printf("Receiving message...n"); pthread_join(client_th, NULL); /* sync */ } } void *client_cb(void *arg_list) { int *ad = (int *)arg_list; /* connected socket descriptor */ char buffer[BUF_SIZE]; int buffer_len, i; if (recv(*ad, buffer, BUF_SIZE, 0) == -1) err_fatal(errno); printf("Message received from client side: %sn", buffer); buffer_len = strlen(buffer); for (i = 0; i < buffer_len; i++) buffer[i] = toupper(buffer[i]); if (send(*ad, buffer, buffer_len, 0) == -1) err_fatal(errno); printf("Transmition finished!n"); if (close(*ad) == -1) err_fatal(errno); }
I have included that defines some makros and a function to handle errors:
#include <stdio.h> #include <string.h> #include <unistd.h> #define BUF_SIZE 512 #define PORT 9000 #define BUFLEN 512 #define NPACK 10 void err_fatal(const int err) { char* msg = strerror(err); printf("%s", msg); exit(-1); }
send and recv system calls
send() function initiates the transmission of a message to its peer. The socket must be connected before a message can be sent. This is true when a server calls send() on an accepted socket and when a client calls send() on a connected socket.
The synopsis is defined as follows:
#include <sys/socket.h> ssize_t send(int socket, const void *buffer, size_t length, int flags);
Here, socket is the accepted or connected socket of the peer to send the message to. buffer contains the message to send and length specifies its size. To specify the type of transmission, you can set flags to one of the following constants that can be logically OR’ed:
MSG_EOR- Terminates a record (if supported by the protocol).
MSG_OOB- Sends out-of-band data on sockets that support out-of-band communications. The significance and semantics of out-of-band data are protocol-specific.
Usually, for simple writing of messages to a socket you can pass 0 (zero).
The recv() function reads a message from a connected socket. That’s all. It’s defined as follows:
#include <sys/socket.h> ssize_t recv(int socket, void *buffer, size_t length, int flags);
Again, socket is the connected socket to read from and buffer points to a memory location to write the message in which is length bytes large. The flags parameter has a different meaning here:
MSG_PEEK- Peeks at an incoming message. The data is treated as unread and the next
recv()or similar function shall still return this data. MSG_OOB- Requests out-of-band data. The significance and semantics of out-of-band data are protocol-specific.
MSG_WAITALL- On SOCK_STREAM sockets this requests that the function block until the full amount of data can be returned. The function may return the smaller amount of data if the socket is a message-based socket, if a signal is caught, if the connection is terminated, if
MSG_PEEKwas specified, or if an error is pending for the socket.
Simply pass 0 (zero) if you don’t know what the above constants mean.
send and recv or read and write?
It really does not matter what you use to read or write from and to a socket. With the flag options you have the chance to apply special options to the communications, but unless you don’t know how to work with OOB or why and when to peek a message, you will be quite happy with read and write either.
Writing a UDP server and client
UDP is a connectionless protocol, that is, it’s not guaranteed in any way that the data send to a peer arrives or that a peer is notified of an error. It is up to the application the perform error checking.
Basically, a few differences to TCP server and client design are worth to be pointed out:
- A UDP server does only call
socketandbind, but notlistenandaccept. It’s reasonable because with UDP there nothing like a connection.listenandacceptrequire a connection to listen at.
A UDP client calls socket but not connect for the same reason.
All we need to do is call socket, and bind if working on server code, and then write to or read from the socket. This is done with the recvfrom and sendto system calls.
recvfrom is defined as follows:
#include <sys/socket.h> ssize_t recvfrom(int socket, void *buffer, size_t length, int flags, struct sockaddr *address, socklen_t *address_len);
Not very surprisingly socket is the socket of the peer we want to receive a message from, buffer points to a memory location suitable to store the messsage in, length specifies the length of the buffer, flags indicate the type of transmission, address points to a properly configured socket address structure and address_len speficies the length of the structure. The flags argument has exactly the same semantics as seen for recv.
sendto is defined as follows:
#include <sys/socket.h> ssize_t sendto(int socket, const void *message, size_t length, int flags, const struct sockaddr *dest_addr, socklen_t dest_len);
Again socket is the socket of the peer the message is to be sent to, message points to a memory location containing the message to be sent, length specifies the size of the message buffer, flags indicates the type of transmission, dest_addr points to the properly configured socket address structure of the peer and dest_len specifies its length.
Let’s put things together and write a client and a server.
UDP server code
#include <stdio.h> #include <sys/types.h> #include <sys/socket.h> #include <netinet/in.h> #include <netdb.h> #include <sys/param.h> #include <errno.h> #include "header.h" int main(void) { struct sockaddr_in local, peer; int s, i, slen = sizeof(peer); char buf[BUFLEN]; if ((s = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP)) < 0) err_fatal(errno); memset((char *)&local, sizeof(local), 0); local.sin_family = AF_INET; local.sin_port = htons(PORT); local.sin_addr.s_addr = htonl(INADDR_ANY); if (bind(s, &local, sizeof(local)) < 0) err_fatal(errno); for (i = 0; i < NPACK; i++) { if (recvfrom(s, buf, BUFLEN, 0, &peer, &slen) < 0) err_fatal(errno); printf("Received packet from %s:%dnData: %snn", inet_ntoa(peer.sin_addr), ntohs(peer.sin_port), buf); } close(s); return 0; }
First, we declare receive buffer and create a socket. AF_INET says that it will be an Internet socket. SOCK_DGRAM says that it will use datagram delivery instead of virtual circuits (another term for connection-oriented delivery mode). IPPROTO_UDP says that it will use the UDP protocol (the standard transport layer protocol for datagrams in IP networks). Generally you can use zero for the last parameter; the kernel will figure out what protocol to use (in this case, it would choose IPPROTO_UDP anyway). As with the TCP server, our UDP server will accept datagrams from all interfaces specified by INADDR_ANY. Now we are ready to bind the socket to the address we created above. This line tells the system that the socket s should be bound to the address in local.
Note that recvfrom() will set slen to the number of bytes actually stored. If you want to play safe, set slen to sizeof(peer) after each call to recvfrom(). The information about the sender we got from recvfrom() is displayed (IP:port), along with the data in the packet. inet_ntoa() takes a struct in_addr and converts it to a string in dot notation, which is rather useful if you want to display the address in a legible form.
UDP client code
#include <stdio.h> #include <sys/types.h> #include <sys/socket.h> #include <netinet/in.h> #include <netdb.h> #include <sys/param.h> #include <errno.h> #include "header.h" int main(int argc, char **argv) { struct sockaddr_in peer; int s, i, slen = sizeof(peer); char buf[BUFLEN]; if (argc != 2) { printf("usage: %s n", argv[0]); exit(0); } if ((s = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP)) < 0) err_fatal(errno); memset((char *)&peer, sizeof(peer), 0); peer.sin_family = AF_INET; peer.sin_port = htons(PORT); if (inet_aton(SRV_IP, &peer.sin_addr) == 0) err_fatal(errno); for (i = 0; i < NPACK; i++) { printf("Sending packet %dn", i); sprintf(buf, "This is packet %dn", i); if (sendto(s, buf, BUFLEN, 0, &peer, slen) < 0) err_fatal(errno); } close(s); return 0; }
The call to inet_aton makes sure that a valid IP address has been given. It will return an error if the address cannot be converted properly. Optionally, You may call bind() after the call to socket(), if you wish to specify which port and interface that should be used for the client socket. However, this is almost never necessary and not really recommended. The system will decide what port and interface to use. Finally, we send BUFLEN bytes from buf to s, with no flags. The receiver is specified in peer, which contains slen byte.
I have not yet decided what to write about in the next article. If you have comments or suggestions on the current article, please contact me. Thanks for reading.
graegerts