Java Thread Ping Pong example

Question

I'm trying to understand thread basics, and as a first example I create two thread that write a String on the stdout. As I know the scheduler allows to execute the threads using a round robin schedule. Thats why I got:

PING PING pong pong pong PING PING PING pong pong

Now I want to use a shared variable, so every thread will know if its your turn:

public class PingPongThread extends Thread {
private String msg;
private static String turn;

public PingPongThread(String msg){
    this.msg = msg;
}
@Override
public void run() {
    while(true) {
        playTurn();
    }

}
public synchronized void playTurn(){
    if (!msg.equals(turn)){
        turn=msg;
        System.out.println(msg);
    }
}
}

Main class:

public class ThreadTest {
    public static void main(String[] args) {
        PingPongThread thread1 = new PingPongThread("PING");
        PingPongThread thread2 = new PingPongThread("pong");
        thread1.start();
        thread2.start();
    }
}

I synchronized the "turn manager" but I still get something like:

PING PING pong pong pong PING PING PING pong pong

Can someone explains what I am missing, and Why I'm not getting Ping pong... ping pong. Thanks!

Answer 1

In conclusion to my discussion with Brian Agnew, I submit this code that uses java.util.concurrent.Phaser to coordinate your ping-pong threads:

static final Phaser p = new Phaser(1);
public static void main(String[] args) {
  t("ping");
  t("pong");
}
private static void t(final String msg) {
  new Thread() { public void run() {
    while (true) {
      System.out.println(msg);
      p.awaitAdvance(p.arrive()+1);
    }
  }}.start();
}

The key difference between this solution and the one you attempted to code is that your solution busy-checks a flag, thereby wasting CPU time (and energy!). The correct approach is to use blocking methods that put a thread to sleep until it is notified of the relevant event.

Answer 2

This line:

public synchronized void playTurn(){
    //code
}

is equivalent in behavior to

public void playTurn() {
    synchronized(this) {
         //code
    }
}

that's why no synchronization is occuring, because as Brian Agnew pointed out, the threads are synchronizing on two different objects (thread1, thread2), each on it's own instance resulting in no effective synchronization.

If you would use your turn variable for synchronization, e.g.:

private static String turn = ""; // must initialize or you ll get an NPE

public void playTurn() {
    synchronized(turn) {
         //...
         turn = msg; // (1)
         //...
    }
}

then the situation is a lot better (run multiple times to verify), but there is also no 100% synchronization. In the beggining (mostly) you get a double ping and double pong, and afterwards they look synchronized, but you still can get double pings/pongs.

The synchronized block locks upon value (see this great answer) and not the reference to that value. (see EDIT)

So let's take a look at one possible scenario:

thread1 locks on ""
thread2 blocks on ""
thread1 changes the value of turn variable to "PING" - thread2 can continue since "" is no longer locked 

To verify that I tried putting

try {
    Thread.currentThread().sleep(1000); // try with 10, 100 also multiple times
 } 
 catch (InterruptedException ex) {}

before and after

turn = msg;

and it looks synchronized?! But, if you put

 try {
    Thread.yield();
    Thread.currentThread().sleep(1000); //  also  try multiple times
 } 
 catch (InterruptedException ex) {}

after few seconds you'll see double pings/pongs. Thread.yield() essenitally means "I'm done with the processor, put some else thread to work". This is obviously system thread scheduler implementation on my OS.

So, to synchronize correctly we must remove line

    turn = msg;

so that threads could always synchronize on the same value - not really :) As explained in the great answer given above - Strings (immutable objects) are dangerous as locks - because if you create String "A" on 100 places in your program all 100 references(variables) will point to the same "A" in memory - so you could oversynchronize.

So, to answer your original question, modify your code like this:

 public void playTurn() {
    synchronized(PingPongThread.class) {
         //code
    }
}

and the parallel PingPong example will be 100% correctly implemented (see EDIT^2).

The above code is equivalent to:

 public static synchronized void playTurn() {
     //code
 }

The PingPongThread.class is a Class object, e.g. on every instance you can call getClass() which always has only one instance.

Also you could do like this

 public static Object lock = new Object();

 public void playTurn() {
    synchronized(lock) {
         //code
    }
}

Also, read and program examples(running multiple times whenever neccessary) this tutorial.

EDIT:

To be technically correct:

The synchronized method is the same as synchronized statement locking upon this. Let's call the argument of the synchronized statement "lock" - as Marko pointed out, "lock" is a variable storing a reference to an object/instance of a class. To quote the spec:

The synchronized statement computes a reference to an object; it then attempts to perform a lock action on that object's monitor..

So the synchronizaton is not acutally made upon value - the object/class instance, but upon the object monitor associated with that instance/value. Because

Each object in Java is associated with a monitor..

the effect remains the same.

EDIT^2:

Following up on the comments remarks: "and the parallel PingPong example will be 100% correctly implemented" - meaning, the desired behavior is achieved (without error).

IMHO, a solution is correct if the result is correct. There are many ways of solving the problem, so the next criteria would be simplicity/elegance of the solution - the phaser solution is better approach, because as Marko said in other words in some comment there is a lot lesser chance of making error using phaser object than using synchronized mechanism - which can be seen from all the (non)solution variants in this post. Notable to see is also comparison of code size and overall clarity.

To conclude, this sort of constructs should be used whenever they are applicable to the problem in question.

Answer 3

Each instance of the PingPongThread is synchronising on itself and not on a shared resource. In order to control the message passing you'll need to synchronise on a shared resource (e.g. your turn variable ?)

However I don't think this is really going to work. I think you should check out wait() and notify() to do this (if you want to understand the threading primitives). See this for an example.

Answer 4

My solution is this :

public class InfinitePingPong extends Thread  {

    private static final Object lock= new Object();

private String toPrintOut;

    public InfinitePingPong(String s){
        this.toPrintOut = s;
    }


    public void run(){
        while (true){
            synchronized(lock){
                System.out.println(this.toPrintOut +" -->"+this.getId()); 
                lock.notifyAll();

                try {
                    lock.wait();
                } catch (InterruptedException e) {}
            }
        }
    }

    public static void main(String[] args) throws InterruptedException {


        InfinitePingPong a = new InfinitePingPong("ping");
        InfinitePingPong b = new InfinitePingPong("pong");


        a.start();
        b.start();

        b.wait();

        try {
            a.join();
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }








}
}

Answer 5

One option is using SynchronousQueue .

import java.util.concurrent.SynchronousQueue;

public class PingPongPattern {

    private SynchronousQueue<Integer> q = new SynchronousQueue<Integer>();
    private Thread t1 = new Thread() {

        @Override
        public void run() {
            while (true) {

                // TODO Auto-generated method stub
                super.run();
                try {

                    System.out.println("Ping");
                    q.put(1);
                    q.put(2);
                } catch (Exception e) {

                }
            }
        }

    };

    private Thread t2 = new Thread() {

        @Override
        public void run() {

            while (true) {
                // TODO Auto-generated method stub
                super.run();
                try {
                    q.take();
                    System.out.println("Pong");
                    q.take();

                } catch (Exception e) {

                }

            }

        }

    };

    public static void main(String[] args) {
        // TODO Auto-generated method stub
        PingPongPattern p = new PingPongPattern();
        p.t1.start();
        p.t2.start();
    }
}

Answer 6

Here is a Ping Pong program written in Java. Ping and Pong are separate threads. Each thread is both a consumer and a producer. When each thread runs it does two things

1. Produce a message that allows the other (as a consumer) to run
2. Consume a message that causes itself to suspend.

The code is based upon Oracles ProducerConsumerExample. Note that the Ping and Pong classes are almost identical in their code and in their behaviour. The threads in the OP’s code only uses the ‘mutual exclusion’ part of the objects monitor (as Brian Agnew suggested above). They never invoke a wait. Hence they only exclude one another, but never invoke the java run time to allow the other thread to run.

/*
 * Copyright (c) 1995, 2008, Oracle and/or its affiliates. All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions
 * are met:
 *
 *   - Redistributions of source code must retain the above copyright
 *     notice, this list of conditions and the following disclaimer.
 *
 *   - Redistributions in binary form must reproduce the above copyright
 *     notice, this list of conditions and the following disclaimer in the
 *     documentation and/or other materials provided with the distribution.
 *
 *   - Neither the name of Oracle or the names of its
 *     contributors may be used to endorse or promote products derived
 *     from this software without specific prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS
 * IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO,
 * THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
 * PURPOSE ARE DISCLAIMED.  IN NO EVENT SHALL THE COPYRIGHT OWNER OR
 * CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
 * EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
 * PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
 * PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
 * LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
 * NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
 * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

 * based on oracle example on sync-wait-notify
 * cf. https://docs.oracle.com/javase/tutorial/essential/concurrency/guardmeth.html
 * run with java ProducerConsumerExample
 * 
 *
 */ 

public class ProducerConsumerExample {
    public static void main(String[] args) {
        Drop drop = new Drop();
    DropCtoP dropCtoP = new DropCtoP();
    (new Thread(new Ping(drop,dropCtoP))).start();
        (new Thread(new Pong(drop,dropCtoP))).start();
    }
}


public class Pong implements Runnable {
    private Drop drop;
    private DropCtoP dropCtoP;
    private int count=0;

    public Pong(Drop drop,DropCtoP dropCtoP) {
        this.drop = drop;
        this.dropCtoP = dropCtoP;
    }

    public void run() {
        String message;
        for (;;) {
        count++;
            message = drop.take();
            System.out.format("Pong running - : %s - ran num times %d %n", message,count);
            dropCtoP.put("Run ping token");
        }
    }
}



public class Ping implements Runnable {
    private Drop drop;
    private DropCtoP dropCtoP;
    private int count=0;

    public Ping(Drop drop,DropCtoP dropCtoP) {
        this.drop = drop;
        this.dropCtoP = dropCtoP;
    }

    public void run() {

        String message;
        for (;;) {
      count++;
      drop.put("Run pong token");
      message = dropCtoP.take();
      System.out.format("PING running - : %s- ran num times %d %n", message,count);
        }

    }
}



public class DropCtoP {
    // Message sent from producer
    // to consumer.
    private String message;
    // True if consumer should wait
    // for producer to send message,
    // false if producer should wait for
    // consumer to retrieve message.
    private boolean empty2 = true;


    public synchronized String take() {
        // Wait until message is
        // available.
        while (empty2) {
            try {
                wait();
            } catch (InterruptedException e) {}
        }
        // Toggle status.
        empty2 = true;
        // Notify producer that
        // status has changed.
        notifyAll();
        return message;
    }

    public synchronized void put(String message) {
        // Wait until message has
        // been retrieved.
        while (!empty2) {
            try { 
                wait();
            } catch (InterruptedException e) {}
        }
        // Toggle status.
        empty2 = false;
        // Store message.
        this.message = message;
        // Notify consumer that status
        // has changed.
        notifyAll();
    }    
}


public class Drop {
    // Message sent from producer
    // to consumer.
    private String message;
    // True if consumer should wait
    // for producer to send message,
    // false if producer should wait for
    // consumer to retrieve message.
    private boolean empty = true;

    public synchronized String take() {
        // Wait until message is
        // available.
        while (empty) {
            try {
                wait();
            } catch (InterruptedException e) {}
        }
        // Toggle status.
        empty = true;
        // Notify producer that
        // status has changed.
        notifyAll();
        return message;
    }

    public synchronized void put(String message) {
        // Wait until message has
        // been retrieved.
        while (!empty) {
            try { 
                wait();
            } catch (InterruptedException e) {}
        }
        // Toggle status.
        empty = false;
        // Store message.
        this.message = message;
        // Notify consumer that status
        // has changed.
        notifyAll();
    }


}

Answer 7

One of the possible implementations:

public class PingPongDemo {

    private static final int THREADS = 2;

    private static int nextIndex = 0;

    private static String getMessage(int index) {
        return index % 2 == 0 ? "ping" : "pong";
    }

    public static void main(String[] args) throws Throwable {
        var lock = new ReentrantLock();

        var conditions = new Condition[THREADS];
        for (int i = 0; i < conditions.length; i++) {
            conditions[i] = lock.newCondition();
        }

        for (int i = 0; i < THREADS; i++) {
            var index = i;

            new Thread(() -> {
                lock.lock();
                try {
                    while (true) {
                        System.out.println(getMessage(index));
                        try {
                            Thread.sleep(1000);
                        } catch (InterruptedException e) {
                            throw new RuntimeException(e);
                        }

                        nextIndex = (nextIndex + 1) % THREADS;

                        conditions[nextIndex].signal();

                        while (nextIndex != index) {
                            conditions[index].awaitUninterruptibly();
                        }
                    }
                } finally {
                    lock.unlock();
                }
            }).start();

            if (index < THREADS - 1) {
                lock.lock();
                try {
                    while (nextIndex != (index + 1)) {
                        conditions[index + 1].awaitUninterruptibly();
                    }
                } finally {
                    lock.unlock();
                }
            }
        }
    }

}

Here, we're effectively making round-robin output.

Answer 8

Here is a version that uses Semaphore objects to accomplish synchronization:

import java.util.concurrent.*;

public class Main {
    @FunctionalInterface
    public interface QuadFunction<T, U, V, W, R> {
        public R apply(T t, U u, V v, W w);
    }

    public static void main(String[] args) {
        ExecutorService svc = Executors.newFixedThreadPool(2);

        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            System.out.println("Terminating...");
            svc.shutdownNow();
            try { svc.awaitTermination(Long.MAX_VALUE, TimeUnit.NANOSECONDS); }
            catch(InterruptedException e) {};
        }));

        var sem1 = new Semaphore(1);
        var sem2 = new Semaphore(0);

        QuadFunction<String, String, Semaphore, Semaphore, Runnable> fun =
            (name, action, s1, s2) ->
                (Runnable) () -> {
                    try {
                        while (true) {
                            s1.acquire();
                            System.out.format("%s %s\n", name, action);
                            Thread.sleep(500);
                            s2.release(1);
                        }
                    } catch (InterruptedException e) {}
                    s2.release(1);
                    System.out.format("==> %s shutdown\n", name);
                };

        svc.execute(fun.apply("T1", "ping", sem1, sem2));
        svc.execute(fun.apply("T2", "pong", sem2, sem1));
    }
}