why use wait/notify over new thread creation?

user_1291986 0
Viewed 172

I did some rough timings of system response times of wait/notify vs creating a new thread over 100 threads and it seems like it is pretty much the same - with the stddev being much higher for wait/notify so seems like creating new threads may be more efficient.

The values are in nano seconds

threads
  {:min 193161, :mean 1410865.0, :stddev 444840.72741115646, :skew -0.755184951510553}

wait/notify
  {:min 159629, :mean 1332270.8, :stddev 703829.1499013959, :skew 0.07768945756036612}

Is it supposed to be more efficient?

Edit The code is in clojure. I'll put some comments in

(def ---thread-benchmark---
  (let [summary (cc/queue)]
    (dotimes [i 100]
      (let [queue (cc/queue)
   
            ;; start time is called
            start (System/nanoTime)
   
            ;; immediately calls future
            _     (future
   
                    ;; and logs the time at which the future is created
                    (let [end  (System/nanoTime)
                          start (cc/take queue)]
                      
                      ;; combines both start and end times
                      (cc/put summary {:start start
                                       :end end
                                       :total (- end start)} )))
            _      (cc/put queue start)]))
    
    ;; wait for threads to complete 
    (Thread/sleep 100)
   
    ;; output results
    (cc/drain summary)))

Wait/Notify Setup:

(defrecord TestStore [instance threads lock])

;; This sets up a store (which has the lock and holds all the threads)

(def -store-
  (let [instance (atom nil)
        threads  (atom {})
        
        ;; This is a single lock
        lock     (Object.)
        watch-fn  (fn [_ _ _ n]
                    (let [;; when watcher is triggered
                          ;; make note of time
                          start (System/nanoTime)
                          
                          ;; notifies all watchers on the lock
                          _ (locking lock
                              (.notifyAll lock))
                          
                          queues (mapv :queue (vals @threads))]
                      
                      ;; send start time to each thread
                      (doseq [q queues]
                        (cc/put q start))))
        _ (add-watch instance :primary watch-fn)]
    (TestStore. instance threads lock)))


;; This holds the queue (which gets sent the results)
(def -results- (atom (cc/queue)))

;; This registers a thread with the store and starts it
(defn register-handler
  [{:keys [instance lock threads] :as store} id f]
  (let [queue  (cc/queue)

        ;;Creates the thread function
        thread-fn (fn []
                    (loop []
                      ;; When it starts, the thread is locked
                      (locking lock (.wait lock))

                      ;; As soon as it get notify, it logs the end time
                      (let [end   (System/nanoTime)
                            start (cc/take queue)]

                        ;; And puts the results on the store
                        (cc/put @-results- {:start start
                                            :end end
                                            :total (- end start)}))
                      (recur)))

        ;; Creates the new Thread
        thread (Thread. thread-fn)

        ;; Registers with the store (so it can send start times)
        _ (swap! threads
                 assoc id {:queue queue
                           :thread thread})]

    ;; starts each thread
    (.start thread)
    store))

Wait/Notify BENCHMARK:

;; REGISTERS 100 THREADS to WAIT on a single lock
(dotimes [i 100]
  (register-handler -store- (str (rand)) (fn [m])))

(def ---wait-notify-benchmark---

  ;; Setup queue (to receive events)
  (let [_ (reset! -results- (cc/queue))]
    
    ;; Triggers the Notify Function 
    (reset! (:instance -store-) 1)

    ;; Wait for results
    (Thread/sleep 100)
    (cc/drain @-results-)))
concurrency clojure java
1 Answers

In my benchmark, creating a new thread each time instead of passing elements to another thread via a Queue and wait()/notify() is 2 orders of magnitude slower.

Either your benchmark is flawed or your scenario is more specific, in which case you should describe it.

If you want to reduce the latency between when an element is posted and when it is consumed, then you need busy-waiting. wait()/notify() and locks call into the kernel which involves some overhead.
You can use JCTools or Agrona.

class WaitNotify {    
    static final int iterations = 10_000_000;

    private static void benchmark() throws InterruptedException {
        Consumer consumer = new Consumer();
        consumer.start();
        long start = System.nanoTime();
        for (int i = 0; i < iterations; i++) {
            synchronized (consumer) {
                consumer.queue.add(new Object());
                consumer.notify();
            }
        }
        consumer.join();
        long elapsed = System.nanoTime() - start;
        System.out.println((double) elapsed / iterations);
    }
}

class Consumer extends Thread {
    final Queue<Object> queue = new ArrayDeque<>();

    @Override
    public synchronized void run() {
        try {
            for (int i = 0; i < WaitNotify.iterations; ) {
                wait();
                while (!queue.isEmpty()) {
                    queue.poll();
                    i++;
                }
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

class NewThread {    
    private static final int iterations = 100_000;

    private static void benchmark() throws InterruptedException {
        long start = System.nanoTime();
        for (int i = 0; i < iterations; i++) {
            Thread consumer = new Consumer(new Object());
            consumer.start();
            consumer.join();
        }
        long elapsed = System.nanoTime() - start;
        System.out.println((double) elapsed / iterations);
    }
}

class Consumer extends Thread {
    Consumer(Object e) {}

    @Override
    public void run() {}
}
user_34532260
Related