                (encode-seq (into-array v) jsonGenerator)))
 
 
-
-(defn -main []
-  (println "launching poller " (System/currentTimeMillis))
-  (let [poll-interval 1000
-        pg-spec         (heroku-db-spec (env/env :database-url))
-        mg-db           (mongo-connect! (env/env :mongo-url))
-        last-timestamp  (atom 0)
-        conversations   (atom {})]
-    (endlessly poll-interval
-      (println "poll >" @last-timestamp)
-      (let [new-votes (poll pg-spec @last-timestamp)
-            zid-to-votes (group-by :zid new-votes)
-            zid-votes (shuffle (into [] zid-to-votes))
-            ]
-        (doseq [[zid votes] zid-votes]
-          (let [lastVoteTimestamp (:created (last votes))]
-            (swap! conversations
-              (fn [convs]
-                (assoc convs zid
-                    (try
-                      (do
-                        (println "zid: " zid)
-                        (conv-update (or (convs zid) {:rating-mat (named-matrix)}) votes))
-                      (catch Exception e
-                        (do
-                          (println "exception when processing zid: " zid)
-                          (.printStackTrace e)))))))
-                
-            (println "zid: " zid)
+(def poll-interval 1000)
+(def pg-spec         (heroku-db-spec (env/env :database-url)))
+(def mg-db           (mongo-connect! (env/env :mongo-url)))
+(def last-timestamp  (atom 0))
+(def conversations   (atom {})) ; generated artifacts
+(def pending-votes   (ref {})) ; zid -> votes
+
+(def dinner-bell (Object.))
+(def active-zids (ref #{}))
+(def pending-votes (ref {})) ; zid -> new votes
+
+(defn notify-workers []
+  (locking dinner-bell
+    (.notifyAll dinner-bell)))
+
+(defn wait-for-dinner-bell []
+  (locking dinner-bell
+    (.wait dinner-bell)))
+
+(defn pending-inactive-zids []
+  (filter (fn [zid] (not (contains? @active-zids zid)))
+          (keys @pending-votes)))
+
+(defn should-take-votes []
+  (> (count (pending-inactive-zids)) 0))
+
+(defn should-not-take-votes []
+  (not (should-take-votes)))
+
+
+(defn make-worker-thread [id]
+ (fn [] ;; closure to capture the id
+  (do
+    (println "starting worker thread" id)
+   (while #(true)
+     (do
+       (println "worker" id "checking for votes")
+    ; if there are votes pending, take a zid and go for it.
+    ; otherwise, wait for a notification
+    (while (should-not-take-votes)
+      (do
+        (println "worker" id "no votes to take")
+        (wait-for-dinner-bell)))
+    ;
+    (let [
+          [zid votes]
+          (dosync
+           ; TODO ensure that we actually get votes here.
+           (let [zid (first (shuffle (pending-inactive-zids)))
+                 votes (@pending-votes zid)]
+             (do
+               (ref-set pending-votes (dissoc @pending-votes zid))
+               (ref-set active-zids (conj @active-zids zid))
+               (vector zid votes))))]
+
+      (println "worker thread" id "starting on math for zid:" zid)
+
+      
+    ; DO MATH
+      (swap!
+       conversations
+       (fn [convs]
+         (assoc convs zid
+                (try
+                  (do
+                    (conv-update (or (convs zid) {:rating-mat (named-matrix)}) votes))
+                  (catch Exception e
+                    (do
+                      (println "exception when processing zid: " zid)
+                      (.printStackTrace e)))))))
+      
+
+    ; UPLOAD/PUBLISH - do this in worker thread so we don't start more
+    ; work on a zid before we upload the result. That may be a later optimization.
+
+      (let [lastVoteTimestamp (:created (last votes))]
+          (println "zid: " zid)
             (println "time: " (System/currentTimeMillis))
             (println "\n\n")
             (let [

            [clojure.java.jdbc :as sql]
            [monger.core :as mg]
            [monger.collection :as mgcol]
            [environ.core :as env]))




                (assoc obj
                 "lastVoteTimestamp" lastVoteTimestamp
                 "zid" zid))))
            )

      (println "worker thread" id "finished zid" zid)
      (dosync ; do-sync needed?
       (ref-set active-zids (disj @active-zids zid)))
  ))))))


(defn merge-pending-votes [zid-votes-pairs]
  (dosync ;; dosync needed?
   (ref-set pending-votes
    (reduce
     (fn [o zid-vote-pair]
       (let [zid (first zid-vote-pair) votes (second zid-vote-pair)]
         (assoc o zid
                (concat (or (get o zid) [])
                        votes))))
     @pending-votes
     zid-votes-pairs
     ))))


(defn -main []
  (println "launching poller " (System/currentTimeMillis))

  (let [threads (map #(Thread. %1)
                     (map make-worker-thread
                          (range 4)))]
    (println (map #(.start %1) threads))
    )


  (endlessly poll-interval
      (println "poll >" @last-timestamp)
      (let [new-votes (poll pg-spec @last-timestamp)
            zid-to-votes (group-by :zid new-votes)
            zid-votes (into [] zid-to-votes)
            ]
        (if (> (count zid-votes) 0)
          (do
            (println "poller found new votes for" (count zid-votes) "conversations.")
            (merge-pending-votes zid-votes)
            (notify-workers)
            (swap! last-timestamp (fn [_] (:created (last new-votes))))))

        )))
