Implementing Monitors, IPC, Semaphore, using each other (CMPT 332)

sync-primitives.c

/* 
 *                GLWT(Good Luck With That) Public License
 *                  Copyright (c) Everyone, except Author
 * 
 * Everyone is permitted to copy, distribute, modify, merge, sell, publish,
 * sublicense or whatever they want with this software but at their OWN RISK.
 * 
 *                             Preamble
 * 
 * The author has absolutely no clue what the code in this project does.
 * It might just work or not, there is no third option.
 * 
 * 
 *                 GOOD LUCK WITH THAT PUBLIC LICENSE
 *    TERMS AND CONDITIONS FOR COPYING, DISTRIBUTION, AND MODIFICATION
 * 
 *   0. You just DO WHATEVER YOU WANT TO as long as you NEVER LEAVE A
 * TRACE TO TRACK THE AUTHOR of the original product to blame for or hold
 * responsible.
 * 
 * IN NO EVENT SHALL THE AUTHORS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
 * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
 * DEALINGS IN THE SOFTWARE.
 * 
 * Good luck and Godspeed.
 *
 */


/*
 * 1. MONITORS WITH IPC 
 * 2. MONITORS WITH SEMAPHORES 
 * 3. SEMAPHORES WITH MONITORS
 * 4. SEMAPHORES WITH IPC
 * 5. IPC WITH MONITORS
 * 6. IPC WITH SEMAPHORES
 */

// ============================================================================
// 1. MONITORS WITH IPC
// ============================================================================

// shared data
LIST *enterq, urgentq;
PID MonitorPid;
int monBusy;
LIST **cvlist;
MonInit (int cv)
{
  enterq = ListCreate ();
  urgentq = ListCreate ();
  for (i = 0; i < cv; i++)
    {
      cvlist[i] = ListCreate ();
    }
  monBusy = 0;
  Create (&MonitorPid, void (*) () Monserver, USR);
}
MonEnter ()
{
  PID id = myPID ();
  MSG msg *;
  msg->pid = &id;
  msg->msgType = ‘e’;
  Send (MonitorPid, &msg, len, reply, len);
}
MonLeave ()
{
  PID id = myPID ();
  MSG msg *;
  msg->pid = &id;
  msg->msgType = ‘l’;
  Send (MonitorPid, &msg, len, reply, len);
}
MonWait (int cv)
{
  PID id = myPID ();
  MSG msg *;
  msg->pid = &id;
  msg.msgType = ‘w’;
  msg->cv = cv;
  Send (MonitorPid, &msg, len, reply, len);
}
MonSignal (int cv)
{
  PID id = myPID ();
  MSG msg *;
  msg->pid = &id;
  msg->msgType = ‘s’;
  msg->cv = cv;
  Send (MonitorPid, &msg, len, reply, len);
}
MonServer ()
{
  MSG msg;
  PID sender;
  PID *qthread;
  while (1)
    {
      Receive (&sender, &msg, len); 
      if (msg.msgType == ‘e’)
      {
        if (monBusy)
          {
            ListPrepend (enterq, msg.pid);
          }
        else
          {
            monBusy = 1;
            Reply (sender, msg, len);
          }
      }
      else if (msg.msgType = ‘l’)
      {
        Reply (sender, msg, len);
        if (ListCount (urgentq) != 0)
          {
            qthread = ListTrim (urgentq);
            Reply (*qthread, msg,
                   len); // reply to both thread leaving and the urgentq thread
          }
        else if (ListCount (enterq) != 0)
          {
            qthread = ListTrim (enterq);
            Reply (*qthread, msg,
                   len); // reply to both thread leaving and the enterq thread
          }
        else
          {
            monBusy = 0;
          }
      }
      else if (msg.msgType == ‘w’)
      {
        ListPrepend (cvlist[msg.cv], sender);
        if (ListCount (urgentq) != 0)
          {
            qthread = ListTrim (urgentq);
            Reply (*qthread, msg,
                   len); // reply to both thread leaving and the urgentq thread
          }
        else if (ListCount (enterq) != 0)
          {
            qthread = ListTrim (enterq);
            Reply (*qthread, msg,
                   len); // reply to both thread leaving and the enterq thread
          }
        else
          {
            monBusy = 0;
          }
      }
      else if (msg.msgType == ‘s’)
      {
if( ListCount(cvlist[msg.cv] != 0){
          ListPrepend (urgentq, sender);
          qthread = ListTrim (cvlist[msg.cv]);
          Reply (*qthread, msg, len);
}
else{
          Reply (sender, msg, len);
}
      }
    }

// ============================================================================
// 2. MONITORS WITH SEMAPHORES
// ============================================================================

// shared data 
Sem enter, urgent, mutex; // fifo sems
LIST *enterq, urgentq;
LIST **cvlist; // list of cv lists
Sem *semcv;
MonInit (int cv)
{
  enter = 1;
  urgent = 0;
  mutex = 1;
  enterq = ListCreate ();
  urgentq = ListCreate ();
  for (i = 0; i < cv; i++)
    {
      semcv[i] = 0;
      cvlist[i] = ListCreate ();
    }
}
MonEnter ()
{
  PID id = myPID ();
  P (mutex)
  ListPrepend (enterq, id);
  V (mutex);
  P (enter);
}
MonLeave ()
{
  if (ListCount (urgentq) != 0)
    {
      P (mutex);
      ListTrim (urgentq);
      V (mutex);
      V (urgent);
    }
  else
    {
      V (enter);
    }
  P (mutex)
  ListTrim (enterq);
  V (mutex);
}
MonWait (int cv)
{
  PID id = myPID ();
  P (mutex);
  ListPrepend (cvlist[cv], &id);
  V (mutex);
  if (ListCount (urgentq) != 0)
    {
      P (mutex);
      ListTrim (urgentq);
      V (mutex);
      V (urgent);
    }
  else
    {
      V (enter);
    }
  P (semcv[cv]);
}
MonSignal (int cv)
{
  PID id = myPID ();
( ListCount(cvlist[cv] != 0){
    P (mutex);
    ListPrepend (urgentq, &id);
    ListTrim (cvlist[cv]);
    V (mutex);
    V (semcv[cv]);
    P (urgent);

}

// ============================================================================
// 3. SEMAPHORES WITH MONITORS
// ============================================================================

typedef struct
{
  int value;         // Current semaphore value
  condition notZero; // Condition for blocking when value is 0
  Monitor *mon;      // Monitor for synchronization
} SemWithMon;

void SemInit (SemWithMon * sem)
{
  // Initialize monitor and condition variable
  sem->mon = MonitorCreate ();
  sem->value = 0;
  sem->notZero = ConditionCreate ();
}

void P (SemWithMon * sem)
{
  MonEnter (sem->mon);

  // If semaphore is 0, wait until it becomes non-zero
  while (sem->value == 0)
    {
      MonWait (sem->mon, sem->notZero);
    }
  sem->value++;

  MonLeave (sem->mon);
}

void V (SemWithMon * sem)
{
  MonEnter (sem->mon);

  sem->value = 0;
  // Signal one waiting thread if any
  MonSignal (sem->mon, sem->notZero);

  MonLeave (sem->mon);
}

// ============================================================================
// 4. SEMAPHORES WITH IPC
// ============================================================================

typedef struct
{
  PID serverPid;   // Server process ID
  int value;       // Semaphore value
  LIST *waitQueue; // Queue of waiting processes
} SemWithIPC;

typedef struct
{
  char type;  // 'P' or 'V'
  PID sender; // Sender's PID
} SemMessage;

void SemServer (SemWithIPC * sem)
{
  SemMessage msg;
  PID sender;

  while (1)
    {
      Receive (&sender, &msg, sizeof (msg));

      if (msg.type == 'P')
        {
          if (sem->value > 0)
            {
              // Resource available
              sem->value--;
              Reply (sender, "OK", 2);
            }
          else
            {
              // Add to wait queue
              ListPrepend (sem->waitQueue, &sender);
              // No reply - sender blocks
            }
        }
      else if (msg.type == 'V')
        {
          sem->value++;
          if (!ListEmpty (sem->waitQueue))
            {
              // Wake up one waiting process
              PID *waiter = ListTrim (sem->waitQueue);
              Reply (*waiter, "OK", 2);
            }
          Reply (sender, "OK", 2);
        }
    }
}

void P (SemWithIPC * sem)
{
  SemMessage msg;
  char reply[2];

  msg.type = 'P';
  msg.sender = myPID ();
  Send (sem->serverPid, &msg, sizeof (msg), reply, 2);
}

void V (SemWithIPC * sem)
{
  SemMessage msg;
  char reply[2];

  msg.type = 'V';
  msg.sender = myPID ();
  Send (sem->serverPid, &msg, sizeof (msg), reply, 2);
}

// ============================================================================
// 5. IPC WITH MONITORS
// ============================================================================

typedef struct
{
  Monitor *mon;         // Monitor for synchronization
  LIST *msgQueue;       // Queue of pending messages
  condition msgAvail;   // Condition for blocking receivers
  condition replyAvail; // Condition for blocking senders
} IPCWithMon;

void Send (IPCWithMon * ipc, PID dest, void *msg, int len)
{
  MonEnter (ipc->mon);

  // Add message to queue
  Message *m = CreateMessage (msg, len, myPID (), dest);
  ListPrepend (ipc->msgQueue, m);

  // Signal any waiting receivers
  MonSignal (ipc->mon, ipc->msgAvail);

  // Wait for reply
  MonWait (ipc->mon, ipc->replyAvail);

  MonLeave (ipc->mon);
}

void Receive (IPCWithMon * ipc, PID * sender, void *msg, int len)
{
  MonEnter (ipc->mon);

  // Wait for message if queue empty
  while (ListEmpty (ipc->msgQueue))
    {
      MonWait (ipc->mon, ipc->msgAvail);
    }

  // Get message from queue
  Message *m = ListTrim (ipc->msgQueue);
  memcpy (msg, m->data, min (len, m->len));
  *sender = m->sender;

  MonLeave (ipc->mon);
}

void Reply (IPCWithMon * ipc, PID dest, void *msg, int len)
{
  MonEnter (ipc->mon);

  // Signal waiting sender
  MonSignal (ipc->mon, ipc->replyAvail);

  MonLeave (ipc->mon);
}

// ============================================================================
// 6. IPC WITH SEMAPHORES
// ============================================================================

typedef struct
{
  Semaphore *mutex;      // For mutual exclusion
  Semaphore *msgAvail;   // Count of available messages
  Semaphore *replyAvail; // For blocking senders
  LIST *msgQueue;        // Queue of pending messages
} IPCWithSem;

void Send (IPCWithSem * ipc, PID dest, void *msg, int len)
{
  P (ipc->mutex);

  // Add message to queue
  Message *m = CreateMessage (msg, len, myPID (), dest);
  ListPrepend (ipc->msgQueue, m);

  // Signal message available
  V (ipc->msgAvail);

  // Release mutex and wait for reply
  V (ipc->mutex);
  P (ipc->replyAvail);
}

void Receive (IPCWithSem * ipc, PID * sender, void *msg, int len)
{
  // Wait for message to be available
  P (ipc->msgAvail);
  P (ipc->mutex);

  // Get message from queue
  Message *m = ListTrim (ipc->msgQueue);
  memcpy (msg, m->data, min (len, m->len));
  *sender = m->sender;

  V (ipc->mutex);
}

void Reply (IPCWithSem * ipc, PID dest, void *msg, int len)
{
  // Signal waiting sender
  V (ipc->replyAvail);
}