Title: Solution
Course: CPS 261: Advanced Java Concepts ~ DL & MM MASTER

Solution to Simultaneous Read, Compute, and Write

Data Structures


import java.util.*;
class buffType
{
    int BUFF_SIZE = 1024;
    byte[] buffer = new byte[BUFF_SIZE];
    int count;
}
class Llist
{
    LinkedList list = new LinkedList();
    boolean stopFlag = false;
    synchronized void stop()
    {
        stopFlag = true;
        notify();
    }
    synchronized void put(buffType b)
    {
        list.addLast(b);
        notify();
    }
    synchronized buffType get()
    {
        while (list.size() == 0)
        {
            if (stopFlag) return null;
            try
            {
                wait();
            }catch (InterruptedException e){}
            if (stopFlag) return null;
        }
        return (buffType)list.removeFirst();
    }
}

// Circular Queue (sort of customized for this problem)
class Cque
{
    int NUM_BUFFERS = 10;
    buffType[] buffers = new buffType[NUM_BUFFERS] ;
    int count = NUM_BUFFERS;
    int getp=0;
    int putp = 0;

    Cque()
    {
        for (int i=0; i < NUM_BUFFERS; i++)
        {
            buffers[i] = new buffType();
        }
    }

    synchronized void put()
    {
      if (count == 0)notify();
      count += 1;
      // If I were doing a general circular queue, I would
      // need more code here.  But the buffers are pre-allocated
      // and I can only do a put if a buffer was already fetched.
    }

    synchronized buffType get()
    {
        while (count == 0)
        {
            try{
                wait();
            } catch (InterruptedException e) {}
        }
        buffType retval = buffers[getp];
        count -= 1;
        getp += 1;
        if (getp >= NUM_BUFFERS) getp = 0;
        return retval;
    }


}


3 Threads


import java.io.*;

class Reader extends Thread
{
    Llist list;
    Cque cque;
    BufferedInputStream binput;
    boolean debug;
    Reader(String filename, Cque cque, Llist list, boolean debug)
         throws IOException
    {
        this.debug = debug;
        this.cque = cque;
        this.list = list;
        FileInputStream fi = new FileInputStream(filename);
        binput = new BufferedInputStream(fi);
    }
    public void run()
    {
      try
      {
        for(;;)
        {
            buffType buff = cque.get();
            int numBytes = binput.read(buff.buffer);
            if (numBytes <= 0)
                break;
            if (debug)
              System.out.println("Reader read another buffer");
            buff.count = numBytes;
            list.put(buff);
        }
        System.out.println("Reader Terminating");
        binput.close();
        list.stop();
       }
       catch (IOException e)
       {
            System.out.println("Reader err: "+e);
       }
    }
}

class ComputeThread extends Thread
{
    Llist in;
    Llist out;
    boolean debug;
    ComputeThread(Llist inList, Llist outList, boolean debug)
    {
        this.debug = debug;
        this.in = inList;
        this.out = outList;
    }
    public void run()
    {
        for(;;)
        {
            buffType buff = in.get();
            if (buff == null) break;
            int count = buff.count;
            byte[] buffer = buff.buffer;
            for (int i=0; i < count; i ++)
            {
                byte b = buffer[i];
                char c = (char)b;
                c = Character.toUpperCase(c);
                buffer[i] = (byte)c;
                // Lets burn up a lot of CPU
                for (int j=0; j< 1000; j++)
                    Math.sqrt(12345.98);
            }
            if (debug)
               System.out.println("Compute Thread processed buffer");
            out.put(buff);
        }
        System.out.println("Compute Thread terminating");
        out.stop();
    }
}

class Writer extends Thread
{
    Llist list;
    Cque cque;
    BufferedOutputStream boutput;
    DoneObject done;
    boolean debug;
    Writer(String filename, Llist list, Cque cque,
         DoneObject done, boolean debug)
         throws IOException
    {
        this.debug = debug;
        this.cque = cque;
        this.list = list;
        this.done = done;
        FileOutputStream fo = new FileOutputStream(filename);
        boutput = new BufferedOutputStream(fo);
    }
    public void run()
    {
      try
      {
        for(;;)
        {
            buffType buff = list.get();
            if (buff == null) break;
            boutput.write(buff.buffer, 0, buff.count);
            if (debug)
               System.out.println("Writer processed a buffer");
            cque.put();
        }
        System.out.println("Writer terminating");
        boutput.close();
        done.setDone();
      }
      catch (IOException e)
      {
        System.out.println("Writer err: "+e);
      }
    }
}

// The sole purpose of this class is to provide
// the main routine a way to know when everything
// has completed.

class DoneObject
{
    synchronized void setDone()
    {
        notify();
    }
    synchronized void waitForDone()
    {
        try
        {
            wait();
        }catch (InterruptedException e){}
    }
}


class TestT
{
    Cque cque = new Cque();
    Llist list1 = new Llist();
    Llist list2 = new Llist();
    DoneObject done = new DoneObject();
    Reader reader ;
    ComputeThread compute;
    Writer writer;

    void doit(String fileIn, String fileOut, boolean debug)
        throws IOException
    {
        reader = new Reader(fileIn, cque, list1, debug);
        compute = new ComputeThread(list1, list2, debug);
        writer = new Writer(fileOut, list2, cque, done, debug);
        writer.start();
        compute.start();
        reader.start();
        done.waitForDone();

    }

    public static void main(String[] args) throws IOException
    {
        if (args.length < 2)
        {
            System.out.println("Useage new 2 filenames");
            return;
        }

        TestT t = new TestT();
        boolean debug=false;
        if (args.length >=3) debug=true;

        long time1 = System.currentTimeMillis();
        t.doit(args[0], args[1], debug);
        long time2 = System.currentTimeMillis();
        System.out.println("Execution Time = " +
            (time2 - time1));

    }
}


// December 2, 2003 1:14 AM