My application needs to be able to help Rim wireless devices ( if you have not heard about thesw wireless devices, have a look at Research In Motion ) to update data to a Linux database server. The Rim device that I developed the software is Rim 957, using Mobitex network. The Linux server will exchange packets with these devices thru a gateway called Cingular IAS ( Internet Access Server ) server, using tcpip packets. This server serves as a gateway between customer server and the Rim devices. The basic data unit that the Rim devices send and receive over the air call mpak ( Mobitex packet ). From my Linux server to IAS server and vice versa, these mpaks are embedded into tcp data stream, using tcp connection.
At the time I were developing this app, the Cingular trasmitter could emit only 1 mpak/s for the development server and 5 mpak/s for the production server. With this slow speed, I decided to build protocol layers that are similar to "reliable UDP". It means this is a connectionless protocol, with the aid of datagram numbering and resending.
In order to present about thread programming and tcpip programming and how to deal with the mpak packets, I would like to present here the lowest layer: the layer that exchange tcpip packets.
1. Notes about pthread programming
Let's talk about thread first. My fist function for this layer is ( don't care about the detail of tcpip stuff
in this function and other functions right now ). Have a quick look at Listing 1
1.int StartThreadTcpConnection( char * pServerName, unsigned int port )
2.{
3.char message[128];
4.pthread_attr_t pthread_attr;
5.struct hostent *ph;
6. client_fd = -1;
7. tcpconnection_on = -1;
8. pthread_cond_init( &tcpconnection_cond, NULL );
9. pthread_mutex_init( &mutex_tcpconnection_cond, NULL );
10.
11. iport = port;
12. ph = gethostbyname( pServerName );
13. if( ph == NULL )
14. {
15. writelog_tcpconnectionlayer( "StartThreadTcpConnection stops: invalid hostname");
16. return -1;
17. }
18. memcpy( network_addr, ph->h_addr, 4 ); // already in network-byte-order
19. pthread_attr_init( &pthread_attr );
20. if( pthread_create( &tcpconnection_tid, &pthread_attr, TcpConnectionProc, NULL ) != 0 )
21. {
22. writelog_tcpconnectionlayer( "StartThreadTcpConnection stops with pthread_create error");
23. return -1;
24. }
25. while(pid == 0 ) usleep(1000*50); /* 50 ms */
26. sprintf( message, "ThreadTcpConnection starts, tid=%ld, pid=%d", tcpconnection_tid, pid );
27. writelog_tcpconnectionlayer(message);
28. return 1;
29.}
Create thread First of all, we can see there are new type definitions such as pthread_t, pthread_attr_t ...These types and function prototypes are defined in pthread.h. The first prameter pthread _t is the thread id that is assigned to each thread being created. pthread _attr_t is the attribute of the thread. We will have a little more explanation on this later. The third parameter of pthread_create is the heart of the thread : the thread procedure. This is where you decide what job(s) this thread must do. In our case, managing the connection with IAS server. The last parameter is the parameter transferring to the thread procedure when it starts. This function will return 0 if a thread is created sucessfully, and the thread id is stored in parameter thread_tid. Returning a non-zero value means this thread is failed to create.
Pthread Attribute
Thread attributes affect to how threads work and their relationship with the system.
Detached/Joinable: if a thread is detached, all resources ( thread descriptor and stack )
that the system allocate to create and to control this thread will be freed automatically when this
thread terminates. In contrary to Detach is Joinable. If a thread is joinable, then its resource
will not be reclaimed automatically unless there is another thread join the former, using
int pthread_join( thread_t thread_tid, void ** thread_return )( This function returns 0
for success) If the thread specifies by thread_tid returns anything, it will be stored in the location specified
by thread_return. To set the thread attribute to Detach or Joinable, we can set the attribute variable using
int pthread_attr_setdetachstate(pthread_attr_t *attr, int detachstate)
before calling pthread_create. Parameter detachstate can be PTHREAD_CREATE_DETACHED
or PTHREAD_CREATE_JOINABLE.
By default, pthread_attr_init will set the attribute to Joinable. For a joinable thread, we can
set it to Detach by calling int pthread_detach( pthread_t pthread_tid ).
This function can be called by any thread.
Then, when will we use detached thread or the other ? If no thread minds of when a thread finishes, then the later
can be a detached.
Schedule Policy: There are three policies: SCHED_OTHER (regular, non-realtime scheduling), SCHED_RR (realtime, round-robin) or SCHED_FIFO (realtime, first-in first-out). To set schedule policy, use int pthread_attr_setschedpolicy(pthread_attr_t *attr, int policy) View the man page of sched_setscheduler for more information of these policies on Linux. The default value for pthread attribute 's policy is SCHED_OTHER.
There are more on thread attributes, but it is out of the scope of this small note.
So far, the explanation perhaps is enough for the two lines 19 and 20 of the Listing 1 : they create a thread that is joinable, and with schedule policy of SCHED_OTHER.
Now we continue to the thread procedure
void * TcpConnectionProc( void * lParam )
{
char message[512];
struct sockaddr_in server_addr;
fd_set write_fd;
pid = getpid();
while(1)
{
pthread_testcancel();
if( tcpconnection_on == 0 ) // still have connection ?
{
//writelog_tcpconnectionlayer( "Connection is on\n");
sleep(2);
continue;
}
writelog_tcpconnectionlayer("TcpConnectionProc detects no connection");
// case that tcpconnection_on = -1 : InIASLayer sets it to -1
memset( &server_addr, 0 , sizeof(server_addr));
server_addr.sin_family = AF_INET;
server_addr.sin_port = htons( iport );
server_addr.sin_addr = *(( in_addr *)network_addr);
// there alredy a previous socket ?, if so ,close it.
if( client_fd != -1 )
close( client_fd );
// create a socket
client_fd = socket( AF_INET, SOCK_STREAM, 0 );
// set asynchronous mode
int socketflags = fcntl( client_fd, F_GETFL, 0);
fcntl( client_fd, F_SETFL, socketflags | O_NONBLOCK );
// prepare for 'select'
FD_ZERO( &write_fd );
FD_SET( client_fd, &write_fd);
// check the connection status here
pthread_mutex_lock( &mutex_tcpconnection_cond);
tcpconnection_on = connect( client_fd, (struct sockaddr *) &server_addr, sizeof(struct sockaddr));
if( select( client_fd+1, NULL, &write_fd, NULL, NULL ) == -1 )
{
sprintf( message,"TcpConnectionProc gets 'select' error: %s", strerror(errno) );
writelog_tcpconnectionlayer( message );
pthread_mutex_unlock( &mutex_tcpconnection_cond);
continue;
}
int size = sizeof(tcpconnection_on);
getsockopt( client_fd, SOL_SOCKET, SO_ERROR, &tcpconnection_on, (socklen_t*)&size);
pthread_mutex_unlock( &mutex_tcpconnection_cond);
if( tcpconnection_on != 0 )
{
writelog_tcpconnectionlayer( "TcpConnectionProc can not make connection. Will try again in 1 second");
sleep(1);
}
else
{
pthread_cond_signal( &tcpconnection_cond);
writelog_tcpconnectionlayer("TcpConnectionProc sucessfully makes tcp connection to Tcp server");
}
}
}