remoteclassv7.java

solve readers-writers problem with RMI

/*
################################################################################
# remoteClassv7.java
#
# Copyright (C) 2007 Fernando G. Tinetti
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA
################################################################################
 *
 *   This is "Version 7 server" i.e.: 
 *      Fix a "low probability error" for version 6, since on accessReader()
 *      after the code (lines 81-98) 
 *                    synchronized (this)
 *                    {
 *                      ...
 *                        waitingProcesses.addElement(waitingreader);
 *                      ...
 *                    }
 *
 *      another thread could execute on leaveWriter() (line 199)
 *
 *                    synchronized(waitingProcesses.firstElement()) { waitingProcesses.firstElement().notify(); } // Wake up a reader
 *      
 *      and the reader is not really waiting on the corresponding monitor 
 *      object. This is because monitors are not semaphores (ok, already 
 *      known). Thus, semaphores should be used, since they hold "permits"
 *      instead of being "mutexed" by the JVM as monitors are. 
 *      In this version, the vector of waiting processes holds semaphores 
 *      instead of objects (monitors) thus when a process calls acquire() 
 *      instead of wait(), the semaphore has "memory" if a previous 
 *      release() has been made, and this is not the behaviour of notify(), 
 *      since a notify() whithout a previous release does not imply anything.
 *      Why is it possible that another thread execute on leaveWriter? Well,
 *      it's because synchronized (this) releases the monitor and other
 *      process is now allowed to continue/enter the "synchronized" code of
 *      this object. 
 *   Used from startService, which creates and registers one object of 
 *   this class.
 */

import java.rmi.RemoteException;               // For rmi
import java.rmi.server.UnicastRemoteObject;    // For rmi

import java.util.Vector;                       // Guess what

import java.lang.Thread;                       // For reporting thread ids

import java.util.concurrent.Semaphore;         // guess what


class RClass extends Semaphore                 // Semaphore for reader processes
{
  protected RClass(int i)
  { // Init a semaphore with i permits
    super(i);
  }
}

class WClass extends Semaphore                 // Semaphore for writer processes
{
  protected WClass(int i)
  { // Init a semaphore with i permits
    super(i);
  }
}


public class remoteClassv7 extends UnicastRemoteObject implements remoteMethod
{

  private int calls;                // Just to count the number of rmi(nvocations)

  private int readers, writers;     // Number of each kind of process currently accessing
  private int wwait;                // Number of writer processes currently waiting

  private Vector<Object> waitingProcesses; // Vector of semaphores each holding one waiting process


  protected remoteClassv7() throws RemoteException 
  {
    super();
    calls   = 0;                              // no service provided so far
    readers = 0;                              // no readers accessing
    writers = 0;                              // no writers accessing
    wwait   = 0;                              // no writers waiting access
    waitingProcesses = new Vector<Object>();  // no processes waiting
  }


  // -------------------           accessReader           -------------------
  public Boolean accessReader() throws RemoteException
  {
    RClass   waitingreader = new RClass(0);   // Just to wait if needed
    Boolean  tosleep;

    synchronized (this)
    {
      calls++;

      System.out.println("Entering accessReader, wwait = " + wwait + " calls = " + calls + " thread "  + Thread.currentThread().getId());

      /* No writers accessing and no (previous) writers waiting for access */
      tosleep = (writers > 0) || (wwait > 0);

      if (!tosleep)
      { /* This reader gets the access */
        readers++;
      }
      else
      { /* This reader to the waiting processes queue (the vector) */
        waitingProcesses.addElement(waitingreader);
      }
    }

    if (tosleep)
    {
      try { waitingreader.acquire(); } catch (InterruptedException e) {System.out.println("ERROR");}
    }

    System.out.println("Leaving accessReader, calls = " + calls + " thread "  + Thread.currentThread().getId() + " readers: " + readers + " writers: " + writers);

    return Boolean.TRUE;
  }



  // -------------------           accessWriter           -------------------
  public Boolean accessWriter() throws RemoteException
  {
    WClass   waitingwriter = new WClass(0);   // Just to wait if needed
    Boolean  tosleep;

    synchronized (this)
    {
      calls++;

      System.out.println("Entering accessWriter, calls = " + calls + " thread "  + Thread.currentThread().getId());

      tosleep = ((writers > 0) || (readers > 0));

      if (!tosleep)
      { /* This writer gets the access */
        writers++;    // well, writers = 1;
      }
      else
      { /* This writer to the waiting processes queue (the vector) */
        wwait++;
        waitingProcesses.addElement(waitingwriter);
      }  
    }

    if (tosleep)
    {
      try { waitingwriter.acquire(); } catch (InterruptedException e) {}
    }

    System.out.println("Leaving accessWriter, calls = " + calls + " thread "  + Thread.currentThread().getId() + " readers: " + readers  + " writers: " + writers);

    return Boolean.TRUE;
  }



  // -------------------           leaveReader           -------------------
  public synchronized Boolean leaveReader() throws RemoteException
  {
    WClass waitingwriter;    // Just in case there is a waiting writer

    System.out.println("Entering leaveReader, reader = " + readers + ", wwait = " + wwait + " thread "  + Thread.currentThread().getId());

    readers--;

    if ((readers == 0) && (wwait > 0))
    { /* The first waiting process is a writer that gets the access */
      writers++;
      wwait--;
      waitingwriter = (WClass) waitingProcesses.firstElement();    // To wake up the process
      waitingProcesses.removeElementAt(0);                // The process does not wait anymore
      waitingwriter.release();  // Wake up
    }

    System.out.println("Leaving leaveReader" + " readers: " + readers + " writers: " + writers);
    return Boolean.TRUE;
  }


  // -------------------           leaveWriter           -------------------
  public synchronized Boolean leaveWriter() throws RemoteException
  {
    System.out.println("Entering leaveWriter " + " thread "  + Thread.currentThread().getId());

    // This writer doesn't access anymore
    writers--;

    if ((waitingProcesses.size() > 0) && (waitingProcesses.firstElement() instanceof WClass)) 
    { /* Wake up a writer */
      wwait--;
      writers++;
      ((WClass) waitingProcesses.firstElement()).release();
      waitingProcesses.removeElementAt(0);
    }
    else
    {
      while ((waitingProcesses.size() > 0) && (waitingProcesses.firstElement() instanceof RClass)) 
      { /* Wake up a reader */
        readers++;
        ((RClass)waitingProcesses.firstElement()).release();  // Wake up
        waitingProcesses.removeElementAt(0);
      }
    }

    System.out.println("Leaving leaveWriter" + " readers: " + readers + " writers: " + writers);
    return Boolean.TRUE;
  }
}