I wanted to write some more JNI code and decided to (re)implement a semaphore for MacOSX. In order to assert somehow correctness of my implementation I hereby supply a (multi)consumer/(multi)producer demo.
semaphore_impl.h (autogenerated by javah):
#include <jni.h> #ifndef MACOSX_SEMAPHORE_IMPL_H #define MACOSX_SEMAPHORE_IMPL_H #ifdef __cplusplus extern "C" { #endif JNIEXPORT void JNICALL Java_net_coderodde_util_concurrent_MacOSXSemaphoreImpl_init(JNIEnv*, jobject, jint); JNIEXPORT void JNICALL Java_net_coderodde_util_concurrent_MacOSXSemaphoreImpl_clean(JNIEnv*, jobject); JNIEXPORT void JNICALL Java_net_coderodde_util_concurrent_MacOSXSemaphoreImpl_lock(JNIEnv*, jobject); JNIEXPORT void JNICALL Java_net_coderodde_util_concurrent_MacOSXSemaphoreImpl_unlock(JNIEnv*, jobject); #ifdef __cplusplus } #endif #endif // MACOSX_SEMAPHORE_IMPL_H
semaphore_impl.cpp:
#include "semaphore_impl.h" #include <mach/task.h> #include <mach/semaphore.h> #include <mach/mach.h> static const char* CLASS = "net/coderodde/util/concurrent/MacOSXSemaphoreImpl"; static const char* FIELD = "semaphoreHandle"; static semaphore_t get_semaphore_handle(JNIEnv* env, jobject obj) { jclass clazz = env->FindClass(CLASS); jfieldID fid = env->GetFieldID(clazz, FIELD, "I"); return env->GetIntField(obj, fid); } JNIEXPORT void JNICALL Java_net_coderodde_util_concurrent_MacOSXSemaphoreImpl_init(JNIEnv* env, jobject obj, jint count) { // Create a handle to a semaphore. semaphore_t semaphore; semaphore_create(current_task(), &semaphore, SYNC_POLICY_FIFO, count); // Get to the 'semaphoreHandle' field. const jclass clazz = env->FindClass(CLASS); const jfieldID semaphore_field_id = env->GetFieldID(clazz, FIELD, "I"); // Store the value of the semaphore to the Java semaphore object. env->SetIntField(obj, semaphore_field_id, semaphore); } JNIEXPORT void JNICALL Java_net_coderodde_util_concurrent_MacOSXSemaphoreImpl_clean(JNIEnv* env, jobject obj) { semaphore_destroy(get_semaphore_handle(env, obj), current_task()); } JNIEXPORT void JNICALL Java_net_coderodde_util_concurrent_MacOSXSemaphoreImpl_lock(JNIEnv* env, jobject obj) { semaphore_wait(get_semaphore_handle(env, obj)); } JNIEXPORT void JNICALL Java_net_coderodde_util_concurrent_MacOSXSemaphoreImpl_unlock(JNIEnv* env, jobject obj) { semaphore_signal(get_semaphore_handle(env, obj)); } Makefile:
libsemaphore.jnilib: semaphore_impl.o g++ -dynamiclib -o libsemaphore.jnilib semaphore_impl.o cp libsemaphore.jnilib /usr/local/lib semaphore_impl.o: semaphore_impl.cpp g++ -std=c++11 -O3 -I/Developer/SDKs/MacOSX10.6.sdk/System/Library/Frameworks/JavaVM.framework/Versions/A/Headers/ -c semaphore_impl.cpp MacOSXSemaphoreImpl.java:
package net.coderodde.util.concurrent; import java.io.File; /** * This class implements the semaphore type interfacing with MacOSX. * * @author Rodion "rodde" Efremov * @version 1.6 */ final class MacOSXSemaphoreImpl implements SemaphoreImpl { static { try { // Putting the .jnilib into working directory is // a safe choice. 'System.getProperty("user.dir")'. System.loadLibrary("semaphore"); } catch (UnsatisfiedLinkError error) { error.printStackTrace(); System.exit(1); } } /** * Holds the handle to a semaphore. */ private int semaphoreHandle; /** * Constructs a semaphore with <code>counter</code> permits. * * @param counter the amount of permits. */ MacOSXSemaphoreImpl(int counter) { init(counter); } /** * Creates the semaphore and loads its handle into <code>semaphoreId</code>. */ @Override public native void init(int counter); /** * Releases resources. */ @Override public native void clean(); /** * Acquires this semaphore. If the current counter of this semaphore is * zero, the calling thread is blocked. */ @Override public native void lock(); /** * Releases this semaphore. Effectively increments the counter of this * semaphore so that other threads may acquire this semaphore. */ @Override public native void unlock(); /** * Release the resources associated with this semaphore. */ @Override protected void finalize() { try { super.finalize(); } catch (Throwable t) {} clean(); } } SemaphoreImpl.java:
package net.coderodde.util.concurrent; /** * This package-private interface defines the API for semaphore implementation * types. * * @author Rodion "rodde" Efremov * @version 1.6 */ interface SemaphoreImpl { /** * Initialization routine. * * @param counter the amount of threads that can pass without blocking. */ void init(int counter); /** * Releases all the resources. */ void clean(); /** * Locks the implementing semaphore. */ void lock(); /** * Unlocks the implementing semaphore. */ void unlock(); } Semaphore.java:
package net.coderodde.util.concurrent; /** * This class implements a semaphore type ported to MacOSX. * * @author Rodion "rodde" Efremov * @version 1.6 */ public class Semaphore { /** * Holds the actual native implementation of this semaphore. */ private final SemaphoreImpl impl; /** * Constructs a new semaphore with <code>counter</code> permits. * * @param counter the amount of threads that can lock this semaphore without * blocking. */ public Semaphore(int counter) { checkCounter(counter); this.impl = new MacOSXSemaphoreImpl(counter); } /** * Acquires this semaphore. */ public void lock() { impl.lock(); } /** * Releases this semaphore. */ public void unlock() { impl.unlock(); } /** * Checks the sanity of <code>counter</code>. * * @param counter the initial amount of permits. */ private static void checkCounter(int counter) { if (counter < 0) { throw new IllegalArgumentException( "The semaphore counter too small: " + counter + ", " + "should be at least 0."); } } } Demo.java:
package net.coderodde.util.concurrent; import java.util.Collections; import java.util.HashSet; import java.util.Random; import java.util.Set; /** * This class implements a demonstration for the semaphore. * * @author Rodion "rodde" Efremov * @version 1.6 */ public class Demo { /** * This character denotes so called "poison pill" for communicating to the * consumers that they should exit. */ private static final Character TERMINATION_SENTINEL = '\u2622'; /** * Implements a consumer thread. */ static class Consumer extends Thread { /** * The buffer to consume from. */ private final ConcurrentBuffer<Character> buffer; /** * Constructs this consumer thread. * * @param buffer the concurrent buffer to consume from. * @param id the ID of this consumer thread. */ Consumer(ConcurrentBuffer<Character> buffer, int id) { this.buffer = buffer; this.setName("Consumer " + id); } /** * The actual code of this consumer thread. */ @Override public void run() { for (;;) { final Character c = buffer.remove(); if (c.equals(TERMINATION_SENTINEL)) { // We have a poison pill. Put it back in the buffer and // terminate this thread. buffer.add(c); return; } } } } /** * This class implements producer threads. */ static class Producer extends Thread { /** * The concurrent set holding all active producers. */ private Set<Producer> activeProducers; /** * The actual concurrent buffer to produce to. */ private final ConcurrentBuffer<Character> buffer; /** * Constructs this producer thread. * * @param buffer the buffer for producing items. * @param id the ID of this producer thread. */ Producer(ConcurrentBuffer<Character> buffer, int id) { this.buffer = buffer; this.setName("Producer " + id); } /** * Sets the set of active producer threads. * * @param set a set of threads. */ void setProducerSet(Set<Producer> set) { activeProducers = set; } /** * The actual code for this producer thread. */ @Override public void run() { final Random rnd = new Random(); for (int i = 0; i < 50; ++i) { final Character c = (char)('A' + rnd.nextInt(26)); buffer.add(c); } activeProducers.remove(this); if (activeProducers.isEmpty()) { // The last thread terminates the consumers. buffer.add(TERMINATION_SENTINEL); } } } /** * Implements a concurrent buffer queue. * * @param <E> the actual type of elements. */ static class ConcurrentBuffer<E> { /** * The default capacity of this buffer. */ private static final int DEFAULT_CAPACITY = 20; /** * A binary semaphore (mutex) for synchronizing the access to internals * of this buffer. */ private final Semaphore mutex; /** * Guards against the empty buffer. */ private final Semaphore fillCount; /** * Guards against the full buffer. */ private final Semaphore emptyCount; /** * The actual storage array. */ private final Object[] storage; /** * The index of the head element. */ private int index; /** * The size of this buffer. */ private int size; /** * Constructs this buffer. */ ConcurrentBuffer() { this.mutex = new Semaphore(1); this.fillCount = new Semaphore(0); this.emptyCount = new Semaphore(DEFAULT_CAPACITY); this.storage = new Object[DEFAULT_CAPACITY]; } /** * Appends <code>element</code> to the tail of this buffer. If this * buffer is full, blocks the calling thread until some space becomes * available. * * @param element the element to append. */ void add(E element) { emptyCount.lock(); mutex.lock(); storage[(index + size) % storage.length] = element; ++size; System.out.println(Thread.currentThread().getName() + " produced " + element + ": " + this); mutex.unlock(); fillCount.unlock(); } /** * Removes the element at the head of this buffer. If this buffer is * empty, blocks the calling thread until some content appears in this * buffer. * * @return the element at the head of this buffer. */ E remove() { fillCount.lock(); mutex.lock(); final E ret = (E) storage[index % storage.length]; index = (index + 1) % storage.length; --size; System.out.println(Thread.currentThread().getName() + " consumed " + ret + ": " + this); mutex.unlock(); emptyCount.unlock(); return ret; } /** * Returns the string representation of the contents of this buffer. * This method is not synchronized. * * @return a string. */ @Override public String toString() { final StringBuilder sb = new StringBuilder("["); for (int i = index, j = 0; j < size; ++j, i = (i + 1) % storage.length) { sb.append(storage[i]); if (j < size - 1) { sb.append(", "); } } return sb.append("]").toString(); } } /** * Implements a demonstration. * * @param consumerAmount the amount of consumers to use. * @param producerAmount the amount of producers to use. */ public static void run(int consumerAmount, int producerAmount) { final Consumer[] consumers = new Consumer[consumerAmount]; final Producer[] producers = new Producer[producerAmount]; final ConcurrentBuffer<Character> buffer = new ConcurrentBuffer<>(); for (int i = 0; i < consumerAmount; ++i) { consumers[i] = new Consumer(buffer, i); } for (int i = 0; i < producerAmount; ++i) { producers[i] = new Producer(buffer, i); } final Set<Producer> producerSet = new HashSet<>(producerAmount); for (final Producer p : producers) { producerSet.add(p); } final Set<Producer> synchronizedSet = Collections.synchronizedSet(producerSet); for (final Producer p : producers) { p.setProducerSet(synchronizedSet); } for (final Producer p : producers) { p.start(); } for (final Consumer c : consumers) { c.start(); } } public static void main(String... args) { run(2, 3); } } So, what do you think?
