What is wrong with the my semaphore application?

Question

The program's aim is to simulate that multiple users add a number to the buffer from 0 to n. Then print the sum of numbers in the buffer. When I run the program, it seems that the threads never end. However, the thread will finish when I run the program in debug mode of Idea and step line by line. Also, I do not exactly know where I need to use my semaphore method P() and V() for mutual exclusion.

Version: JDK 8. I cannot use semaphore in the library.

Main.java

Buffer b = new Buffer(bufferSize);                      
ArrayList<user> us = new ArrayList<>();
for(int i = 0; i < num_users; i++) us.add(new user(i, elements, b));
ArrayList<Thread> th = new ArrayList<>();
for(int i = 0; i < num_users; i++)
{
    th.add(new Thread(us.get(i)));
    th.get(i).start();
}

user.java

public class user implements Runnable
{                               
    private int id;
    private int num_elements;
    private semaphore mutex = new semaphore(1 );
    public static Buffer buf;

    public user(int i, int el, Buffer b)                        
        {id = i; num_elements = el; buf = b;}

    public void add_elements()
    {//Add element to buffer, element value iterates from 0, 1, 2 .... num_elements
        mutex.P();
        int n = 0;
        while (num_elements > 0)
        {                       
            buf.add(new Integer(n));                            
            n++;
            num_elements--;
        }
        mutex.V();
    }

    public void run()
    {
        add_elements();
    }
}

Buffer.java

public class Buffer
{                                       
    private LinkedList<Object> buf_list;        
    private int elements;                       //Number of elements currently on the queue
    private int buf_size;                       //Maximum number of elements allowed on queue
    private semaphore mutex = new semaphore(1);

    public Buffer(int n)                        //Queue creation, with n indicating the maximum capacity
    {
        buf_list = new LinkedList<Object>();
        elements = 0;
        buf_size = n;
    }

    public void add(Integer n)
    {       
        mutex.P();
        buf_list.add(n);
        elements++;
        mutex.V();
    }

    public void finalSummation()                
    {                       
        if (elements == buf_size)
        {
            mutex.P();
            int sum = 0;
            for (Object n : buf_list)
                sum += ((Integer)n).intValue();
            mutex.V();
            System.out.println("Count total: " + sum); 
        }
    }
}

semaphore.java

public class semaphore
{
    private int count = 0;

    public semaphore(int init_count)
    {
        count = init_count;
    }

    public synchronized void P()
    {
        count -= 1;
        while (count < 0)
        {
            try { 
                wait();
            } catch (InterruptedException e)  {
                System.out.println("Error");
                System.exit(-1);
            }
        }
    }

    public synchronized void V()
    {
        count += 1;
        notifyAll();
    }
}   

I expect it will print the sum of buffer numbers, but the thread may not finish.

Answer 1

I have put it together using the above answer if that helps. Please select above answer as the right answer.

import java.util.ArrayList;
import java.util.LinkedList;

public class Main {

    // assumed values
    private static final int bufferSize = 5;
    private static final int num_users = 10;
    private static final int elements = 5;

    public static void main(String[] args) {

        Buffer b = new Buffer(bufferSize);                      
        ArrayList<User> us = new ArrayList<>();

        for(int i = 0; i < num_users; i++) us.add(new User(i, elements, b));
        ArrayList<Thread> th = new ArrayList<>();
        for(int i = 0; i < num_users; i++)
        {
            th.add(new Thread(us.get(i)));
            th.get(i).start();
        }       
        for(int i = 0; i < num_users; i++) {
            try {
                th.get(i).join();
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }
        b.finalSummation();
        System.out.println("Exiting");

    }

}

class User implements Runnable
{                               
    private int id;
    private int num_elements;
    public static Buffer buf;

    public User(int i, int el, Buffer b)                        
        {id = i; num_elements = el; buf = b;}

    public void add_elements()
    {//Add element to buffer, element value iterates from 0, 1, 2 .... num_elements
        int n = 0;
        while (num_elements > 0)
        {                       
            buf.add(new Integer(n));                            
            n++;
            num_elements--;
        }
    }

    public void run()
    {
        add_elements();
    }
}

class Buffer
{                                       
    private LinkedList<Object> buf_list;        
    private int elements;                       //Number of elements currently on the queue
    private int buf_size;                       //Maximum number of elements allowed on queue
    private Semaphore mutex ;

    public Buffer(int n)                        //Queue creation, with n indicating the maximum capacity
    {
        buf_list = new LinkedList<Object>();
        elements = 0;
        buf_size = n;
        mutex = new Semaphore(buf_size);
    }

    public synchronized void  add(Integer n)
    {       
        mutex.acquire();
        buf_list.add(n);
        elements++;
        mutex.release();
    }

    public  void finalSummation()                
    {                       
        int sum = 0;
        System.out.println(buf_list);
        for (Object n : buf_list)
            sum += ((Integer)n).intValue();
        System.out.println("Count total: " + sum); 
    }
}

class Semaphore {
    private int permits = 0;

    public Semaphore(int permits) {
        this.permits = permits;
    }

    public synchronized void acquire() {
        while (permits < 1) {
            try {
                wait();
            } catch (InterruptedException e) {
                System.out.println("Error");
                System.exit(-1);
            }
        }
        permits--;
    }

    public void release() {
        synchronized (this) {
            permits += 1;
            notifyAll();
        }
    }
}

Answer 2

There are a few things that stand out as issues here.

1) your code is never calling the finalSummation method. So the "printing" of the result will never happen.

2) Buffer and each user are all creating their own semaphores. If you are attempting to allow multiple threads to update Buffer without colliding then you need to share the same semaphore. Remove the semaphore and the usage of it from the user class. Just let the Buffer instance control only one update at a time with its semaphore.

3) You don't need to check the semaphore in the finalSummation method. Presumably, all threads are done at that point. And to enforce that ...

4) Put code like this at the end of main

for(int i = 0; i < num_users; i++) {
    th.get(i).join();
}
b.finalSummation();

5) A semaphore should manage a number of permits. Your semaphore is managing the number of instances waiting - that is a pretty much an irrelevant number for a semaphore. Change your P() and V() to acquire() and release() to be consistent with the pattern.

public static class semaphore {
    private int permits = 0;

    public semaphore(int permits) {
        this.permits = permits;
    }

    public synchronized void acquire() {
        while (permits < 1) {
            try {
                wait();
            } catch (InterruptedException e) {
                System.out.println("Error");
                System.exit(-1);
            }
        }
        permits--;
    }

    public void release() {
        synchronized (this) {
            permits += 1;
            notifyAll();
        }
    }
}