1. How do I achieve synchronization?
a) By using a semaphore:
In java, you can use counting semaphore as a mutex by setting "available number" to be one.
i.e.
Semaphore available = new Semaphore(1);
//entering critical_section
available.acquire();
//doing what you must do and this part should be short. Be careful, don't do blocking job here, otherwise you are dead.
...
//leaving critical_section
available.release();
b) By using a file lock:
i) How to get?
In java, you can acquire a FileLock from FileChannel.
i.e.
RandomAccessFile myFile=new RandomAccessFile("myFile.txt", "rw");
...
//Be careful! This is a blocking call and you may get deadlock if you don't handle properly
FileLock myLock= myFile.getChannel().lock(); //there are several versions of locking, see below
//You reach here means you already get the file lock and you don't have to worry file may be changed by others.
...
ii) How to use?
In java, filelock is mandatory which means even the owner must "release" filelock before you can modify file.
i.e.
RandomAccessFile myFile=new RandomAccessFile("myFile.txt", "rw");
...
FileLock myLock= myFile.getChannel().lock();
//If you reach here, it means you get the lock. But then why should I release what I just get? This seems to be insane. :)
myLock.release(); //And here you have a synchronization flaw and I leave it to you to solve.
//operating on file
...
iii) Choice: blocking or non-blocking?
Basically it is your design choice to use either blocking or non-blocking call of various "lock" method, but there is some
subtle issue you may encounter in your server implementation. Blocking version (i.e. FileChannel.lock() ) has potential
risk of deadlock and usually has a low throughput. So, I suggest you always choose non-blocking version like "tryLock()".
However, you have to handle the exception in case the file is already locked by others. See explanation of sun.com.
c) By using java key word synchronized to declare functions:
a) "synchronized" key words is class-wise:
In java language, there is a native synchronization mechanism by declaring functions as "synchronized" so that "all" these methods
become "synchronized", i.e. function call becomes "atomic".
class Foo
{
boolean critical;
int synchronized funcA()
{
//by declaring "synchronized", we don't need to worry about the below context switch problem.
if (critical) //this is the risky place because context switch may happen here if we don't declare "synchronized"
return 100
else
return 200;
}
int synchronized funcB()
{
if (critical) //this is the risky place because context switch may happen here if we don't declare "synchronized"
return 300
else
return 400;
}
...
}
Now assume that one object instance of Foo is called by multiple threads then we don't need to do any synchronization at all. Please note not only multiple "funcA" calls are called one by one, also multiple "funcA" and "funcB" calls are also synchronized because "synchronized" key word is for all "synchronized" methods within class Foo! i.e. All "synchronized" methods of class Foo is automatically put in a "logical queue" and is executed exactly one by one no matter how many threads there are.
b) a big trap of "synchronized"
In our project implementation, we once encounter a deadlock scenario by miss-using both synchronized key word because when a method is declared to be synchronized it becomes unable to re-enter. See the following code:
Just imagine that you have two threads running in your server. And one of them is a monitor threads, i.e. TCP
server socket, and the other one is a working thread which is doing some job, i.e. file operation. The monitor
threads is calling "funcA" to check a flag "done" which may be modified by another thread by calling "funcB".
If the status "done" is modified, the monitoring job is finished. This can be very common in both our assignment
and project. However, the subtle thing is that by declaring both "funcA" and "funcB" as synchronized, you are
making them both atomic call. When "threadA" (the monitor thread) is running, "threadB" can never be able to
call "funcB" and you are dead.
import java.io.*;
public class MyClass
{
static boolean done=false;
public static void main(String argv[])
{
MyThread threadA=new MyThread(0);
MyThread threadB=new MyThread(1);
System.out.println(" starts two threads");
threadA.start();
threadB.start();
System.out.println(" two threads started");
}
static synchronized void funcA()
{
try
{
while (!done)
{
System.out.println("done is false and sleep...");
Thread.sleep(1000);
}
}
catch(Exception er)
{
}
}
static synchronized void funcB()
{
System.out.println("I never reach here!");
done=true;
}
static class MyThread extends Thread
{
int myID;
public MyThread(int id)
{
myID=id;
}
public void run()
{
if (myID==0)
{
funcA();
}
else
{
funcB();
}
}
}
}
performance comparison between conditional wait and naive mutex
In linux, there is a superior synchronization tool, namely conditional wait. I design a simple performance test to compare
the performance between naive mutex with conditional wait. The outcome is stunning! Conditional wait finishes instantly
within one second. And naive mutex needs as long as 102seconds. And what's more, I have to force the idle thread "sleep",
otherwise it is not progressing at all!
#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
#include <unistd.h>
#include <sys/time.h>
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
pthread_mutex_t dummy = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t cond = PTHREAD_COND_INITIALIZER;
//void *functionCount1(void*);
//void *functionCount2(void*);
void* condProc(void*param);
void* mutexProc(void*param);
int count = -1;
#define COUNT_DONE 50
//#define COUNT_HALT1 3
//#define COUNT_HALT2 6
typedef void*(*ThreadProc)(void*);
const int MaxThreadNumber=10;
pthread_t threads[MaxThreadNumber];
struct timeval start, end;
//int threadID[MaxThreadNumber];
//ThreadProc threadProc=myProc;
int i;
float result;
int totalSec, totalUsec;
void perform_test(ThreadProc proc)
{
count=-1;
float tempStart, tempEnd;
int usec, sec;
pthread_mutex_init(&mutex, NULL);
pthread_cond_init(&cond, NULL);
for (i=0; i<MaxThreadNumber; i++)
{
pthread_create(threads+i,NULL, proc, (void*)&i);
}
gettimeofday(&start, NULL);
count=0;
pthread_cond_broadcast(&cond);
for (i=0; i<MaxThreadNumber; i++)
{
pthread_join(threads[i], NULL);
}
gettimeofday(&end, NULL);
if (end.tv_usec>start.tv_usec)
{
usec=end.tv_usec-start.tv_usec;
sec=end.tv_sec-start.tv_sec;
}
else
{
usec=1000000+end.tv_usec-start.tv_usec;
sec=end.tv_sec-1-start.tv_sec;
}
/*
tempEnd=end.tv_sec+(float)end.tv_usec/1000000.0;
tempStart=start.tv_sec+(float)start.tv_usec/1000000.0;
printf("start=%f, end=%f\n", tempStart, tempEnd);
result=tempEnd-tempStart;
printf("the total time is: %f\n", result);
*/
printf("sec=%d, usec=%d\n", sec, usec);
totalSec+=sec;
totalUsec+=usec;
}
int main()
{
float total;
int i;
const int TotalTestNumber=20;
total=0;
totalSec=totalUsec=0;
for (i=0; i<TotalTestNumber; i++)
{
printf("condition [%d]\n", i);
perform_test(condProc);
total+=result;
//printf("condition[%d]: %f\n", i, result);
}
//printf("condition wait is on average of %f\n", total/(float)TotalTestNumber);
printf("condition wait is sec=%d, usec=%d\n", totalSec, totalUsec);
total=0.0;
totalSec=totalUsec=0;
for (i=0; i<TotalTestNumber; i++)
{
printf("mutex[%d]\n", i);
perform_test(mutexProc);
total+=result;
//printf("mutex[%d]: %f\n", i, result);
}
//printf("mutex wait is on average of %f\n", total/(float)TotalTestNumber);
printf("mutex wait is sec=%d, usec=%d\n", totalSec, totalUsec);
return 0;
}
void* condProc(void* param)
{
int index=*((int*)param);
//printf("thread %d starts...\n", index);
pthread_mutex_lock(&mutex);
while (count<0)
{
pthread_cond_wait(&cond, &mutex);
}
pthread_mutex_unlock(&mutex);
pthread_cond_broadcast(&cond);
while (true)
{
pthread_mutex_lock(&mutex);
while (count%MaxThreadNumber!=index&&count<COUNT_DONE)
{
pthread_cond_wait(&cond, &mutex);
}
if (count==COUNT_DONE)
{
pthread_mutex_unlock(&mutex);
pthread_cond_broadcast(&cond);
break;
}
else
{
//printf("thread %d doing work to inc count %d\n", index, count);
count++;
pthread_mutex_unlock(&mutex);
//sleep(rand()%2);
pthread_cond_broadcast(&cond);
}
}
}
void* mutexProc(void* param)
{
int index=*((int*)param);
/*
pthread_mutex_lock(&mutex);
while (count<0)
{
pthread_cond_wait(&cond, &mutex);
}
pthread_mutex_unlock(&mutex);
pthread_cond_broadcast(&cond);
*/
printf("thread %d starts...\n", index);
while (true)
{
//printf("thread %d\n", index);
pthread_mutex_lock(&mutex);
if (count%MaxThreadNumber==index)
{
count++;
}
pthread_mutex_unlock(&mutex);
sleep(1);
if (count==COUNT_DONE)
{
break;
}
}
}
/*
void *functionCount1(void* param)
{
for(;;)
{
sleep(rand()%2);
printf("before mutex lock 1\n");
pthread_mutex_lock( &condition_mutex );
sleep(rand()%2);
while( count%2 ==0 )
{
sleep(rand()%2);
printf("before cond wait 1\n");
pthread_cond_wait( &condition_cond, &condition_mutex );
printf("after cond wait 1\n");
sleep(rand()%2);
}
sleep(rand()%2);
count++;
printf("Counter value functionCount1: %d\n",count);
pthread_mutex_unlock( &condition_mutex );
sleep(rand()%2);
if(count >= COUNT_DONE) return(NULL);
}
}
void *functionCount2(void* param)
{
for(;;)
{
sleep(rand()%2);
printf("before mutex lock 2\n");
pthread_mutex_lock( &condition_mutex );
printf("after mutex lock 2\n");
sleep(rand()%2);
if( count%2!=0 )
{
sleep(rand()%2);
printf("before signal 2\n");
pthread_mutex_unlock( &condition_mutex );
pthread_cond_signal( &condition_cond );
printf("after signal 2\n");
sleep(rand()%2);
}
else
{
count++;
printf("Counter value functionCount2: %d\n",count);
pthread_mutex_unlock( &condition_mutex );
sleep(rand()%2);
}
sleep(rand()%2);
if(count >= COUNT_DONE) return(NULL);
}
}
*/