Thursday, July 24, 2014

DelayQueue Usage Java Concurrency

DelayQueue is an unbounded BlockingQueue of objects that implement the Delayed interface. An object can only be taken from the queue when its delay has expired. The queue is sorted so that the object at the head has a delay that has expired for the longest time. If no delay has expired, then there is no head element and poll() will return null.

You can't place null elements in the queue.

Here's an example where the Delayed objects are themselves tasks, and the DelayedTaskConsumer takes the most "urgent" task (the one that has been expired for the longest time) off the queue and runs it. Note that DelayQueue is thus a variation of a priority queue.

import java.util.concurrent.*;
import java.util.*;

import static java.util.concurrent.TimeUnit.*;
 
class DelayedTask implements Runnable, Delayed {

    private static int counter = 0;
    private final int id = counter++;
    private final int delta;
    private final long trigger;
    protected static List<delayedtask> sequence
            = new ArrayList<delayedtask>();

    public DelayedTask(int delayInMilliseconds) {
        delta = delayInMilliseconds;
        trigger = System.nanoTime()
                + NANOSECONDS.convert(delta, MILLISECONDS);
        sequence.add(this);
    }

    public long getDelay(TimeUnit unit) {
        return unit.convert(
                trigger - System.nanoTime(), NANOSECONDS);
    }

    public int compareTo(Delayed arg) {
        DelayedTask that = (DelayedTask) arg;
        if (trigger < that.trigger) {
            return -1;
        }
        if (trigger > that.trigger) {
            return 1;
        }
        return 0;
    }

    public void run() {
        System.out.println(this + " ");
    }

    public String toString() {
        return String.format("[%1$-4d]", delta) + " Task " + id;
    }

    public String summary() {
        return "(" + id + ":" + delta + ")";
    }

    public static class EndSentinel extends DelayedTask {

        private ExecutorService exec;

        public EndSentinel(int delay, ExecutorService e) {
            super(delay);
            exec = e;

        }

        public void run() {
            for (DelayedTask pt : sequence) {
                System.out.println(pt.summary());    //sequence is a static list
            }
            System.out.println();
            System.out.println(this + " calling shutdownNow()");
            exec.shutdownNow();
        }
    }
}

class DelayedTaskConsumer implements Runnable {

    private DelayQueue<delayedtask> q;

    public DelayedTaskConsumer(DelayQueue<delayedtask> q) {
        this.q = q;
    }

    public void run() {
        try {
            while (!Thread.interrupted()) {
                q.take().run(); // Run task with the current thread
            }
        } catch (InterruptedException e) {
// Acceptable way to exit
        }
        System.out.println("Finished DelayedTaskConsumer");
    }
}

public class DelayQueueDemo {

    public static void main(String[] args) {
        Random rand = new Random(47);
        ExecutorService exec = Executors.newCachedThreadPool();
        DelayQueue<delayedtask> queue
                = new DelayQueue<>();
// Fill with tasks that have random delays:
        for (int i = 0; i < 20; i++) {
            queue.put(new DelayedTask(rand.nextInt(5000)));
        }
// Set the stopping point
        queue.add(new DelayedTask.EndSentinel(5000, exec));
        exec.execute(new DelayedTaskConsumer(queue));
    }
}

Program Output: 

[128 ] Task 11
[200 ] Task 7
[429 ] Task 5
[520 ] Task 18
[555 ] Task 1
[961 ] Task 4
[998 ] Task 16
[1207] Task 9
[1693] Task 2
[1809] Task 14
[1861] Task 3
[2278] Task 15
[3288] Task 10
[3551] Task 12
[4258] Task 0
[4258] Task 19
[4522] Task 8
[4589] Task 13
[4861] Task 17
[4868] Task 6
(0:4258)
(1:555)
(2:1693)
(3:1861)
(4:961)
(5:429)
(6:4868)
(7:200)
(8:4522)
(9:1207)
(10:3288)
(11:128)
(12:3551)
(13:4589)
(14:1809)
(15:2278)
(16:998)
(17:4861)
(18:520)
(19:4258)
(20:5000)

[5000] Task 20 calling shutdownNow()

Finished DelayedTaskConsumer

DelayedTask contains a List called sequence that preserves the order in which the tasks were created, so that we can see that sorting does in fact take place.
The Delayed interface has one method, getDelay( ), which tells how long it is until the delay time expires or how long ago the delay time has expired. This method forces us to use the TimeUnit class because that’s the argument type. This turns out to be a very convenient class because you can easily convert units without doing any calculations. For example, the value of delta is stored in milliseconds, but the Java SE5 method System.nanoTime( )

produces time in nanoseconds. You can convert the value of delta by saying what units it is in and what units you want it to be in, like this:

NANOSECONDS.convert(delta, MILLISECONDS);

In getDelay( ), the desired units are passed in as the unit argument, and you use this to convert the time difference from the trigger time to the units requested by the caller, without even knowing what those units are (this is a simple example of the Strategy design pattern, where part of the algorithm is passed in as an argument).
For sorting, the Delayed interface also inherits the Comparable interface, so compareTo( ) must be implemented so that it produces a reasonable comparison. toString( ) and summary( ) provide output formatting, and the nested EndSentinel class provides a way to shut everything down by placing it as the last element in the queue.
Note that because DelayedTaskConsumer is itself a task, it has its own Thread which it can use to run each task that comes out of the queue. Since the tasks are being performed in queue priority order, there’s no need in this example to start separate threads to run the DelayedTasks.
You can see from the output that the order in which the tasks are created has no effect on execution order—instead, the tasks are executed in delay order as expected.

Source: Thinking In Java 4th Ed (Concurrency)

Monday, July 21, 2014

Difference between Unicast, Broadcast, Multicast and Anycast traffic. TCP IP [Networking]

------------------------------------------------------------
| TYPE      | ASSOCIATIONS     | SCOPE           | EXAMPLE |
------------------------------------------------------------
| Unicast   | 1 to 1           | Whole network   | HTTP    | 
------------------------------------------------------------
| Broadcast | 1 to Many        | Subnet          | ARP     |
------------------------------------------------------------
| Multicast | One/Many to Many | Defined horizon | SLP     |
------------------------------------------------------------
| Anycast   | Many to Few      | Whole network   | 6to4    |
------------------------------------------------------------
Unicast is used when two network nodes need to talk to each other. TCP by definition is a Unicast protocol, except when there is Anycast involved (more on that below).
When you need to have more than two nodes see the traffic, you have options.
If all of the nodes are on the same subnet, then broadcast becomes a viable solution. All nodes on the subnet will see all traffic. There is no TCP-like connection state maintained. Broadcast is a layer 2 feature in the Ethernet protocol, and also a layer 3 feature in IPv4.
Multicast is like a broadcast that can cross subnets, but unlike broadcast does not touch all nodes. Nodes have to subscribe to a multicast group to receive information. Multicast protocols are usually UDP protocols, since by definition no connection-state can be maintained. Nodes transmitting data to a multicast group do not know what nodes are receiving. By default, Internet routers do not pass Multicast traffic. For internal use, though, it is perfectly allowed; thus, "Defined horizon" in the above chart. Multicast is a layer 3 feature of IPv4 & IPv6.
To use anycast you advertise the same network in multiple spots of the Internet, and rely on shortest-path calculations to funnel clients to your multiple locations. As far the network nodes themselves are concerned, they're using a unicast connection to talk to your anycasted nodes. For more on Anycast, try: What is "anycast" and how is it helpful?. Anycast is also a layer 3 feature, but is a function of how route-coalescing happens.

Examples

Some examples of how the non-Unicast methods are used in the real Internet.
Broadcast
ARP is a broadcast protocol, and is used by TCP/IP stacks to determine how to send traffic to other nodes on the network. If the destination is on the same subnet, ARP is used to figure out the MAC address that goes to the stated IP address. This is a Level 2 (Ethernet) broadcast, to the reserved FF:FF:FF:FF:FF:FF MAC address.
Also, Microsoft's machine browsing protocol is famously broadcast based. Work-arounds like WINS were created to allow cross-subnet browsing. This involves a Level 3 (IP) broadcast, which is an IP packet with the Destination address listed as the broadcast address of the subnet (in 192.168.101.0/24, the broadcast address would be 192.168.101.255).
The NTP protocol allows a broadcast method for announcing time sources.
Multicast
Inside a corporate network, Multicast can deliver live video to multiple nodes without having to have massive bandwidth on the part of the server delivering the video feed. This way you can have a video server feeding a 720p stream on only a 100Mb connection, and yet still serve that feed to 3000 clients.
When Novell moved away from IPX and to IP, they had to pick a service-advertising protocol to replace the SAP protocol in IPX. In IPX, the Service Advertising Protocol, did a network-wide announcement every time it announced a service was available. As TCP/IP lacked such a global announcement protocol, Novell chose to use a Multicast based protocol instead: the Service Location Protocol. New servers announce their services on the SLP multicast group. Clients looking for specific types of services announce their need to the multicast group and listen for unicasted replies.
HP printers announce their presence on a multicast group by default. With the right tools, it makes it real easy to learn what printers are available on your network.
The NTP protocol also allows a multicast method (IP 224.0.1.1) for announcing time sources to areas beyond just the one subnet.
Anycast
Anycast is a bit special since Unicast layers on top of it. Anycast is announcing the same network in different parts of the network, in order to decrease the network hops needed to get to that network.
The 6to4 IPv6 transition protocol uses Anycast. 6to4 gateways announce their presence on a specific IP, 192.88.99.1. Clients looking to use a 6to4 gateway send traffic to 192.88.99.1 and trust the network to deliver the connection request to a 6to4 router.
NTP services for especially popular NTP hosts may very well be anycasted, but I don't have proof of this. There is nothing in the protocol to prevent it.
Other services use Anycast to improve data locality to end users. Google does Anycast with its search pages in some places (and geo-IP in others). The Root DNS servers use Anycast for similar reasons. ServerFault itself just might go there, they do have datacenters in New York and Oregon, but hasn't gone there yet.

Thursday, July 17, 2014

Dining Philosophers Problem [Code] : [Java Concurrency]

The dining philosophers problem, invented by Edsger Dijkstra, is the classic demonstration of deadlock. The basic description specifies five philosophers (but the example shown here will allow any number). These philosophers spend part of their time thinking and part of their time eating. While they are thinking, they don’t need any shared resources, but they eat using a limited number of utensils. In the original problem description, the utensils are forks, and two forks are required to get spaghetti from a bowl in the middle of the table, but it seems to make more sense to say that the utensils are chopsticks. Clearly, each philosopher will require two chopsticks in order to eat.
A difficulty is introduced into the problem: As philosophers, they have very little money, so they can only afford five chopsticks (more generally, the same number of chopsticks as philosophers). These are spaced around the table between them. When a philosopher wants to eat, that philosopher must pick up the chopstick to the left and the one to the right. If the philosopher on either side is using a desired chopstick, our philosopher must wait until the necessary chopsticks become available.

The code below has a possibility of deadlock. If every philosopher picks the chopstick to the right of him the last philosopher won't be able to pick his right chopstick, thus no one is able to pick his left and eat food, waiting on each other in a chain to get the second chopstick.

Code below taken from Thinking In Java 4th Edition p1224 onwards]

import java.util.concurrent.*;
import java.util.*;

class Chopstick {

    private boolean taken = false;

    public synchronized
            void take() throws InterruptedException {
        while (taken) {
            wait();
        }
        taken = true;
    }

    public synchronized void drop() {
        taken = false;
        notifyAll();
    }
}

class Philosopher implements Runnable {

    private Chopstick left;
    private Chopstick right;
    private final int id;
    private final int ponderFactor;
    private Random rand = new Random(47);

    private void pause() throws InterruptedException {
        if (ponderFactor == 0) {
            return;
        }
        TimeUnit.MILLISECONDS.sleep(
                rand.nextInt(ponderFactor * 250));
    }

    public Philosopher(Chopstick left, Chopstick right,
            int ident, int ponder) {
        this.left = left;
        this.right = right;
        id = ident;
        ponderFactor = ponder;
    }

    public void run() {
        try {
            while (!Thread.interrupted()) {
                System.out.println(this + " " + "thinking");
                pause();
                // Philosopher becomes hungry
                System.out.println(this + " " + "grabbing right");
                right.take();
                System.out.println(this + " " + "grabbing left");
                left.take();
                System.out.println(this + " " + "eating");
                pause();
                right.drop();
                left.drop();
            }
        } catch (InterruptedException e) {
            System.out.println(this + " " + "exiting via interrupt");
        }
    }

    public String toString() {
        return "Philosopher " + id;
    }
}

public class DeadlockingDiningPhilosophers {

    public static void main(String[] args) throws Exception {
        int ponder = 5;
        if (args.length > 0) {
            ponder = Integer.parseInt(args[0]);
        }
        int size = 5;
        if (args.length > 1) {
            size = Integer.parseInt(args[1]);
        }
        ExecutorService exec = Executors.newCachedThreadPool();
        Chopstick[] sticks = new Chopstick[size];
        for (int i = 0; i < size; i++) {
            sticks[i] = new Chopstick();
        }
        for (int i = 0; i < size; i++) {
            exec.execute(new Philosopher(
                    sticks[i], sticks[(i + 1) % size], i, ponder));
        }
        if (args.length == 3 && args[2].equals("timeout")) {
            TimeUnit.SECONDS.sleep(5);
        } else {
            System.out.println("Press 'Enter' to quit");
            System.in.read();
        }
        exec.shutdownNow();
    }
}

You can set the "ponder" variable to 0 to see the deadlock occuring fast. For a deadlock to occur, the following four conditions must be met: 1. Mutual exclusion. At least one resource used by the tasks must not be shareable. In this case, a Chopstick can be used by only one Philosopher at a time. 2. At least one task must be holding a resource and waiting to acquire a resource currently held by another task. That is, for deadlock to occur, a Philosopher must be holding one Chopstick and waiting for another one. 3. A resource cannot be preemptively taken away from a task. Tasks only release resources as a normal event. Our Philosophers are polite and they don’t grab Chopsticks from other Philosophers. 4. A circular wait can happen, whereby a task waits on a resource held by another task, which in turn is waiting on a resource held by another task, and so on, until one of the tasks is waiting on a resource held by the first task, thus gridlocking everything. In this example, the circular wait happens because each Philosopher tries to get the right Chopstick first and then the left. Now to make the code deadlock free we can make the last philosopher get the left chopstick first and then the right chopstick so that the circular chain is broken. The deadlock never occurs now. There are many other ways for avoiding the deadlock in this case, this is just one of them.

 Code:

import java.util.concurrent.*;
import java.util.*;

class Chopstick {

    private boolean taken = false;

    public synchronized
            void take() throws InterruptedException {
        while (taken) {
            wait();
        }
        taken = true;
    }

    public synchronized void drop() {
        taken = false;
        notifyAll();
    }
}

class Philosopher implements Runnable {

    private Chopstick left;
    private Chopstick right;
    private final int id;
    private final int ponderFactor;
    private Random rand = new Random(47);

    private void pause() throws InterruptedException {
        if (ponderFactor == 0) {
            return;
        }
        TimeUnit.MILLISECONDS.sleep(
                rand.nextInt(ponderFactor * 250));
    }

    public Philosopher(Chopstick left, Chopstick right,
            int ident, int ponder) {
        this.left = left;
        this.right = right;
        id = ident;
        ponderFactor = ponder;
    }

    public void run() {
        try {
            while (!Thread.interrupted()) {
                System.out.println(this + " " + "thinking");
                pause();
// Philosopher becomes hungry
                System.out.println(this + " " + "grabbing right");
                right.take();
                System.out.println(this + " " + "grabbing left");
                left.take();
                System.out.println(this + " " + "eating");
                pause();
                right.drop();
                left.drop();
            }
        } catch (InterruptedException e) {
            System.out.println(this + " " + "exiting via interrupt");
        }
    }

    public String toString() {
        return "Philosopher " + id;
    }
}

public class FixedDiningPhilosophers {

    public static void main(String[] args) throws Exception {
        int ponder = 5;
        if (args.length > 0) {
            ponder = Integer.parseInt(args[0]);
        }
        int size = 5;
        if (args.length > 1) {
            size = Integer.parseInt(args[1]);
        }
        ExecutorService exec = Executors.newCachedThreadPool();
        Chopstick[] sticks = new Chopstick[size];
        for (int i = 0; i < size; i++) {
            sticks[i] = new Chopstick();
        }
        for (int i = 0; i < size; i++) {
            if (i < (size - 1)) {
                exec.execute(new Philosopher(
                        sticks[i], sticks[i + 1], i, ponder));
            } else {
                exec.execute(new Philosopher(
                        sticks[0], sticks[i], i, ponder));
            }
        }
        if (args.length == 3 && args[2].equals("timeout")) {
            TimeUnit.SECONDS.sleep(5);
        } else {
            System.out.println("Press 'Enter' to quit");
            System.in.read();
        }
        exec.shutdownNow();
    }
}

Exercise: Change DeadlockingDiningPhilosophers.java so that when a philosopher is done with its chopsticks, it drops them into a bin. When a philosopher wants to eat, it takes the next two available chopsticks from the bin. Does this eliminate the possibility of deadlock? Can you reintroduce deadlock by simply reducing the number of available chopsticks?
My solution:

import java.util.concurrent.*;
import java.util.*;
import static net.mindview.util.Print.*;

class Chopstick {

    private boolean taken = false;

    public synchronized void take() throws InterruptedException {
        while (taken) {
            wait();
        }
        taken = true;
    }

    public synchronized void drop() {
        taken = false;
        notifyAll();
    }
}

class Bin {

    BlockingQueue bin = new LinkedBlockingQueue<>();

    public void put(Chopstick stick) throws InterruptedException {
        bin.put(stick);
    }

    public Chopstick get() throws InterruptedException {
        return bin.take();
    }
}

class Philosopher implements Runnable {

    private Chopstick left;
    private Chopstick right;
    private LinkedBlockingQueue bin;
    private final int id;
    private final int ponderFactor;
    private Random rand = new Random(47);

    private void pause() throws InterruptedException {
        if (ponderFactor == 0) {
            return;
        }
        TimeUnit.MILLISECONDS.sleep(rand.nextInt(ponderFactor * 250));
    }

    public Philosopher(Chopstick left, Chopstick right,
            LinkedBlockingQueue bin, int ident, int ponder) {
        this.left = left;
        this.right = right;
        this.bin = bin;
        id = ident;
        ponderFactor = ponder;
    }

    public void run() {
        try {
            while (!Thread.interrupted()) {
                print(this + " " + "thinking");
                pause();
// Philosopher becomes hungry
                print(this + " taking first, right chopstick");
                right = bin.take();
                print(this + " taking second, left chopstick");
                left = bin.take();
                print(this + " eating");
                pause();
                print(this + " returning chopsticks");
                bin.put(right);
                bin.put(left);
            }
        } catch (InterruptedException e) {
            print(this + " " + "exiting via interrupt");
        }
    }

    public String toString() {
        return "Philosopher " + id;
    }
}

public class DeadlockingDiningPhilosophers {

    public static void main(String[] args) throws Exception {
        int ponder = 0;
        if (args.length > 0) {
            ponder = Integer.parseInt(args[0]);
        }
        int size = 5;
        if (args.length > 1) {
            size = Integer.parseInt(args[1]);
        }
        ExecutorService exec = Executors.newCachedThreadPool();
// chopstick bin:
        LinkedBlockingQueue bin = new LinkedBlockingQueue<>();
        Chopstick[] sticks = new Chopstick[size];
        for (int i = 0; i < size; i++) {
            sticks[i] = new Chopstick();
            bin.put(sticks[i]);
        }
        for (int i = 0; i < size; i++) {
            exec.execute(new Philosopher(sticks[i], sticks[(i + 1) % size], bin, i, ponder));
        }
        if (args.length == 3 && args[2].equals("timeout")) {
            TimeUnit.SECONDS.sleep(5);
        } else {
            System.out.println("Press 'Enter' to quit");
            System.in.read();
        }
        exec.shutdownNow();
    }
}

Does this eliminate the possibility of deadlock? No. Consider the case when each philosopher takes a single chopstick from the bin so that now the bin contains 0 chopsticks. Nobody has the second chopstick to eat the spaghetti.

Java Concurrency : Usage of BlockingQueue class

BlockingQueue allows you to put objects into it and take out objects one at a time. So you don't need to worry about concurrent threads' synchronization for queue access. If there are no elements, the accessing thread simply blocks and resumes when elements are added for the thread to access.

Example Code:

import java.util.concurrent.*;
import java.io.*;
import java.util.*;

class Sender implements Runnable {

    private Random rand = new Random(47);
    private BlockingQueue bq = new LinkedBlockingQueue<>();

    public BlockingQueue getBlockingQueue() {
        return bq;
    }

    public void run() {
        try {
            while (true) {
                for (char c = 'A'; c <= 'z'; c++) {
                    bq.add(c);
                    TimeUnit.MILLISECONDS.sleep(rand.nextInt(500));
                }
            }
        } catch (InterruptedException e) {
            System.out.println(e + " Sender sleep Interrupted");
        }
    }
}

class Receiver implements Runnable {

    private PipedReader in;
    private BlockingQueue bq;

    public Receiver(Sender sender) throws IOException {
        bq = sender.getBlockingQueue();
    }

    public void run() {
        try {
            while (true) {
                System.out.print("Read: " + (Character) (bq.take()) + " ");
            }
        } catch (InterruptedException ex) {
            System.out.println("Reader Interrupted!");
        }
    }
}

public class BlockingQueueTest {

    public static void main(String[] args) throws Exception {
        Sender sender = new Sender();
        Receiver receiver = new Receiver(sender);
        ExecutorService exec = Executors.newCachedThreadPool();
        exec.execute(sender);
        exec.execute(receiver);
        TimeUnit.SECONDS.sleep(5);
        exec.shutdownNow();
    }
}

Output:

Read: A Read: B Read: C Read: D Read: E Read: F Read: G Read: H Read: I Read: J Read: K Read: L Read: M Read: N Read: O Read: P Read: Q Read: R Read: S Read: T Read: U Reader Interrupted!
java.lang.InterruptedException: sleep interrupted Sender sleep Interrupted

Wednesday, July 16, 2014

Java Multithreading : Using Lock and Condition Objects : Restaurant Simulation Example Code [ Concurrency ]

Consider a restaurant that has one chef and one waitperson. The waitperson must wait for the chef to prepare a meal. When the chef has a meal ready, the chef notifies the waitperson, who then gets and delivers the meal and goes back to waiting. After the meal is delivered, the waitperson should notify the BusBoy (new Class) to clean up. This is an example of task cooperation: The chef represents the producer, and the waitperson represents the consumer. Both tasks must handshake with each other as meals are produced and consumed, and the system must shut down in an orderly fashion. Use explicit Lock and Condition objects. Here is the story modeled in code:

[Note: This is the solution to questions taken from Thinking In Java 4th Edition p1212 and p1215]

Code:

import java.util.concurrent.*;
import java.util.concurrent.locks.*;

class BusBoy implements Runnable {

    Lock lock = new ReentrantLock();
    Condition condition = lock.newCondition();

    Restaurant restaurant;

    BusBoy(Restaurant r) {
        restaurant = r;
    }

    public void run() {
        try {
            while (!Thread.interrupted()) {

                lock.lock();
                try {
                    condition.await();
                    System.out.println("BusBoy is Cleaning up!\n");
                } finally {
                    lock.unlock();
                }
            }
        } catch (InterruptedException e) {
            System.out.println("BusBoy interrupted!");
        }
    }

}

class Meal {

    private final int orderNum;
    volatile int cleanUp = 0;

    public Meal(int orderNum) {
        this.orderNum = orderNum;
    }

    public String toString() {
        return "Meal " + orderNum;
    }
}

class WaitPerson implements Runnable {

    Lock lock = new ReentrantLock();
    Condition condition = lock.newCondition();

    private Restaurant restaurant;

    public WaitPerson(Restaurant r) {
        restaurant = r;
    }

    public void run() {
        try {
            while (!Thread.interrupted()) {
                lock.lock();
                try {
                    while (restaurant.meal == null) {
                        condition.await(); //... for the chef to produce a meal
                    }
                } finally {
                    lock.unlock();
                }
                System.out.println("Waitperson got " + restaurant.meal);

                restaurant.chef.lock.lock();
                try {
                    restaurant.meal = null;
                    System.out.println("Meal taken by the waitperson!");
                    restaurant.chef.condition.signalAll(); //Ready for another
                } finally {
                    restaurant.chef.lock.unlock();
                }

                try {
                    restaurant.boy.lock.lock();
                    System.out.println("Notifying BusBoy to cleanup...");
                    restaurant.boy.condition.signalAll();
                } finally {
                    restaurant.boy.lock.unlock();
                }
            }
        } catch (InterruptedException e) {
            System.out.println("WaitPerson interrupted!");
        }
    }
}

class Chef implements Runnable {

    Lock lock = new ReentrantLock();
    Condition condition = lock.newCondition();
    private Restaurant restaurant;
    private int count = 0;

    public Chef(Restaurant r) {
        restaurant = r;
    }

    public void run() {
        try {
            while (!Thread.interrupted()) {
                lock.lock();
                try {

                    while (restaurant.meal != null) {
                        condition.await();//... for the meal to be taken
                    }
                } finally {
                    lock.unlock();
                }

                if (++count == 10) {
                    System.out.println("Out of food, closing");
                    restaurant.exec.shutdownNow();
                    return;
                }
                System.out.println("Order up!");
                restaurant.waitPerson.lock.lock();
                try {
                    restaurant.meal = new Meal(count);
                    restaurant.waitPerson.condition.signalAll();
                } finally {
                    restaurant.waitPerson.lock.unlock();
                }
                TimeUnit.MILLISECONDS.sleep(100);
            }
        } catch (InterruptedException e) {
            System.out.println("Chef interrupted!");
        }
    }
}

public class Restaurant {

    Meal meal;
    ExecutorService exec = Executors.newCachedThreadPool();
    WaitPerson waitPerson = new WaitPerson(this);
    Chef chef = new Chef(this);
    BusBoy boy = new BusBoy(this);

    public Restaurant() {
        exec.execute(chef);
        exec.execute(waitPerson);
        exec.execute(boy);
    }

    public static void main(String[] args) {
        new Restaurant();
    }

}

Output:

Order up!
Waitperson got Meal 1
Meal taken by the waitperson!
Notifying BusBoy to cleanup...
BusBoy is Cleaning up!

Order up!
Waitperson got Meal 2
Meal taken by the waitperson!
Notifying BusBoy to cleanup...
BusBoy is Cleaning up!

Order up!
Waitperson got Meal 3
Meal taken by the waitperson!
Notifying BusBoy to cleanup...
BusBoy is Cleaning up!

Order up!
Waitperson got Meal 4
Meal taken by the waitperson!
Notifying BusBoy to cleanup...
BusBoy is Cleaning up!

Order up!
Waitperson got Meal 5
Meal taken by the waitperson!
Notifying BusBoy to cleanup...
BusBoy is Cleaning up!

Order up!
Waitperson got Meal 6
Meal taken by the waitperson!
Notifying BusBoy to cleanup...
BusBoy is Cleaning up!

Order up!
Waitperson got Meal 7
Meal taken by the waitperson!
Notifying BusBoy to cleanup...
BusBoy is Cleaning up!

Order up!
Waitperson got Meal 8
Meal taken by the waitperson!
Notifying BusBoy to cleanup...
BusBoy is Cleaning up!

Order up!
Waitperson got Meal 9
Meal taken by the waitperson!
Notifying BusBoy to cleanup...
BusBoy is Cleaning up!

Out of food, closing
WaitPerson interrupted!
BusBoy interrupted!

Implementation of Infinite Buffer Producer-Consumer Problem using Java [Multi-Threading / Concurrency]

Constraints:

1. The producer should not put items into the buffer if the consumer has yet to consume an item to free
the circular buffer.

2. The consumer should not consume an item that has not been produced by the producer yet.

Code:

/*
        Author: Jatin Thakur
                coderbots.blogspot.com
*/
 
import java.util.Arrays;
import java.util.concurrent.*;
 
class Consumer implements Runnable {
 
    private final Scenario scenario;
 
    Consumer(Scenario s) {
        scenario = s;
        Arrays.fill(buffer, 0);
    }
 
    int buffer[] = new int[10];
 
    volatile int current = 0;
    volatile int total = 0;
 
    public void run() {
        try {
            while (!Thread.interrupted()) {
                synchronized (this) {
                    while (buffer[current] == 0) {
                        wait();
                    }
                }
                synchronized (scenario.producer) {
                    System.out.println("Consuming item: " + current);
                    buffer[current++] = 0;
 
                    ++total;
                    scenario.producer.notifyAll();
                }
                if (current == 10) {
                    current = 0;
                }
                Thread.yield();
            }
        } catch (InterruptedException e) {
            System.out.println("Consumer interrupted!");
 
        }
    }
}
 
class Producer implements Runnable {
 
    volatile int total = 0;
    private final Scenario scenario;
 
    Producer(Scenario s) {
        scenario = s;
    }
    volatile int current = 0;
 
    public void run() {
        try {
            while (!Thread.interrupted()) {
                synchronized (this) {
                    while (scenario.consumer.buffer[(current == 9 ? -1 : current) + 1] == 1) {
                        wait();
                    }
                }
                synchronized (scenario.consumer) {
                    System.out.println("Producing item: " + current);
                    scenario.consumer.buffer[current++] = 1;
                    scenario.consumer.notifyAll();
 
                    ++total;
                }
                if (current == 10) {
                    current = 0;
                }
                Thread.yield();
            }
        } catch (InterruptedException e) {
            System.out.println("Producer interrupted!");
        }
    }
}
 
public class Scenario {
 
    Consumer consumer = new Consumer(this);
    Producer producer = new Producer(this);
 
    ExecutorService exec = Executors.newCachedThreadPool();
 
    Scenario() {
        exec.execute(consumer);
        exec.execute(producer);
    }
 
    public static void main(String[] args) throws InterruptedException {
        Scenario sc = new Scenario();
        TimeUnit.MILLISECONDS.sleep(5);
        sc.exec.shutdownNow();
        System.out.println("Total items produced: " + sc.producer.total);
        System.out.println("Total items consumed: " + sc.consumer.total + "\n\n");
    }
}

Sample Run:

Producing item: 0
Producing item: 1
Producing item: 2
Producing item: 3
Producing item: 4
Producing item: 5
Producing item: 6
Producing item: 7
Producing item: 8
Consuming item: 0
Consuming item: 1
Consuming item: 2
Consuming item: 3
Consuming item: 4
Consuming item: 5
Consuming item: 6
Consuming item: 7
Consuming item: 8
Producing item: 9
Producing item: 0
Producing item: 1
Producing item: 2
Producing item: 3
Producing item: 4
Producing item: 5
Producing item: 6
Producing item: 7
Consuming item: 9
Consuming item: 0
Consuming item: 1
Consuming item: 2
Consuming item: 3
Consuming item: 4
Consuming item: 5
Consuming item: 6
Consuming item: 7
Producing item: 8
Producing item: 9
Consuming item: 8
Consuming item: 9
Producing item: 0
Producing item: 1
Producing item: 2
Producing item: 3
Producing item: 4
Producing item: 5
Producing item: 6
Producing item: 7
Producing item: 8
Consuming item: 0
Consuming item: 1
Consuming item: 2
Consuming item: 3
Consuming item: 4
Consuming item: 5
Consuming item: 6
Consuming item: 7
Producing item: 9
Total items produced: 30
Total items consumed: 28

If you find any errors, please do comment.

Monday, July 14, 2014

Program for reducing fractions to lowest terms [Java]

So this is just some random program given as a programming exercise by a friend of mine - Saksham. He is pretty excited about his new website.

The program reduces the terms to their lowest values in a fraction. For example:

Enter numerator:
2500
Enter denominator:
50
Reduced form: 50/1

Enter numerator:
9
Enter denominator:
21
Reduced form: 3/7

Code:

import java.util.Scanner;

public class FractionReducer{
    public static void main(String[] args) {
        Scanner scan = new Scanner(System.in);
        System.out.println("Enter numerator: ");
        int num = scan.nextInt();
        System.out.println("Enter denominator: ");
        int den = scan.nextInt();

        int smaller = num < den ? num : den;
        int HCF = -1;
        for (int i = smaller; i > 0; --i) {
            if (num%i==0&&den%i==0) {
                HCF = i;
                System.out.println("Reduced form: "+(num/HCF)+"/"+(den/HCF));
                break;
            }
        }
    }
}

MultiThreaded Program to Check and Display the Reachable IP addresses on a Network [Java]

This program uses multiple threads to look out for reachable IPs on a network and then displays the IPs that are reachable. You can adjust the timeout in inReachable(timeout) [in milliseconds] and the number of threads in the for loop's check in main function.


import java.io.IOException;
import java.util.concurrent.*;
import java.util.regex.*;
import java.util.*;
import java.net.*;

class Worker implements Runnable {

    static int id = 0;
    final int iden = ++id;
    static volatile String ip = "192.168.80.1";
    static Pattern pattern = Pattern.compile(".\\d+$");
    static Matcher m = null;
    static volatile int i = 1;
    static volatile boolean cancel = false;
    static volatile List<String> reachables = new ArrayList<>();

    public static synchronized void ping(int id, String ipLocal) throws UnknownHostException, IOException {
        InetAddress adr = null;
        if (i < 255) {

            m = pattern.matcher(ip);
            if (m.find()) {
                ip = m.replaceFirst("." + Integer.toString(++i));
                System.out.print("\nThread #" + id + " testing IP: " + ip);

                adr = InetAddress.getByName(ip);
                ipLocal = ip;
            }
        } else {
            cancel = true;
            return;
        }

        if (adr.isReachable(2000)) {
            System.out.print("\nAddress " + ipLocal + " is reachable!");
            reachables.add(ipLocal);
        } else {
            System.out.print("\nAddress " + ipLocal + " not reachable!");
        }
    }

    public void run() {
        String ipLocal = "";
        while (!cancel) {
            try {
                ping(iden, ipLocal);
                Thread.yield();
            } catch (IOException ex) {
                System.out.println("IOException caught!");
            }

        }
    }
}

public class Ping {

    public static void main(String[] args) throws InterruptedException {
        ExecutorService exec = Executors.newCachedThreadPool();

        for (int j = 0; j < 10; ++j) {
            exec.execute(new Worker());
        }

        exec.shutdown();
        while (true) {
            if (exec.isTerminated()) {
                System.out.println("\n\nReachable IPs = ");
                for (String reach : Worker.reachables) {
                    System.out.println(reach);
                }
                break;
            }
        }
    }
}

Sample Output:

Thread #1 testing IP: 192.168.80.2
Address 192.168.80.2 is reachable!
Thread #10 testing IP: 192.168.80.3
Address 192.168.80.3 not reachable!
Thread #10 testing IP: 192.168.80.4
Address 192.168.80.4 not reachable!
Thread #10 testing IP: 192.168.80.5
Address 192.168.80.5 not reachable!
Thread #10 testing IP: 192.168.80.6
Address 192.168.80.6 not reachable!
Thread #10 testing IP: 192.168.80.7
Address 192.168.80.7 is reachable!
Thread #10 testing IP: 192.168.80.8
Address 192.168.80.8 not reachable!


Reachable IPs = 
192.169.80.2
192.168.80.7


Sunday, July 13, 2014

Concurrency: Using wait() and notifyAll() [Example Program][Java]

wait(), notify() and notifyAll() must only be placed within synchronized methods or blocks. sleep() can be called within non-synchronized methods. If you call any of these methods within a method that's not synchronized, the program will compile, but when you run it, you'll get an IllegalMonitorStateException.

1. The object lock is released during the wait().
2. You can also come out of the wait() due to a notify() or notifyAll(), or if the timeout occurs (using the
     timed version of wait    ->    wait(pause).

import java.util.concurrent.*;

class Runnable1 implements Runnable {

    public synchronized void waiter() {
        try {
            wait();
            System.out.println("Runnable1 notified!");
        } catch (InterruptedException ex) {
            System.out.println("Runnable1 interrupted! - Not notified!");
        }
    }

    @Override
    public void run() {
        waiter();
    }
}

class Runnable2 implements Runnable {

    Runnable1 ref;

    Runnable2(Runnable1 run) {
        ref = run;
    }

    public void notifier() {
        synchronized (ref) {
            ref.notifyAll();
        }
    }

    @Override
    public void run() {
        while (!Thread.interrupted()) {
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException ex) {
                System.out.println("Sleep interrupted!");
                break;
            }
             notifier();
        }
        System.out.println("Exiting Runnable2 run()");
    }

}

public class Ex21 {

    public static void main(String[] args) throws InterruptedException {
        ExecutorService exec = Executors.newCachedThreadPool();
        Runnable1 ref1 = new Runnable1();
        exec.execute(ref1);
        exec.execute(new Runnable2(ref1));
        TimeUnit.SECONDS.sleep(2);
        exec.shutdownNow();
        System.out.println("All tasks terminated!");
    }
}

Runnable1 first wait() s to get notified by Runnable2's notifier(). And then displays the message that
it was notified. But if timeout occurs (after calling shutdownNow() on the Executor) (3 seconds / change timeout values to see different ouput), Runnable1 displays the message that it wasn't notified.

Output:

If Runnable1 gets notified:

Runnable1 notified!
Sleep interrupted!
Exiting Runnable2 run()
All tasks terminated!

If it doesn't:

All tasks terminated!
Runnable1 interrupted! - Not notified!
Sleep interrupted!
Exiting Runnable2 run()