                       (cancel-requests rs)
                       nil)
            (= c cl) (recur (inc i))))))))
+
+(defn bring-up-instances [is]
+  (let [n  (count is)
+        id (.toString (UUID/randomUUID))
+        cl (notify-chan id)
+        to (timeout (* 5 60 1000))]
+    (doseq [i is]
+            (modify-instance-attribute :instance-id i :user-data (s->b64 (send-up id))))
+    (start is)
+    (async/go-loop [i n]
+      (if (zero? i)
+        (instance-info-map is)
+        (let [[v c] (async/alts! [cl to])]
+          (cond
+           (= c to) (do (info "Timeout on instance request:" is)
+                        (close-notify-chan! id)
+                        (stop is)
+                        nil)
+           (= c cl) (recur (dec i))))))))
 
 
 (def ag (ssh/ssh-agent {}))

          (>! c req) (close! c)))
    c))

(comment 
(def my-subnets {"us-east-1c"  "subnet-08eff44e"
                 "us-east-1a"  "subnet-f290abda"
                 "us-east-1b"  "subnet-572dd420"})
(def my-zone    "us-east-1a")
(def my-req {:spot-price 		0.01
             :instance-count 		1
             :type 			"one-time"
             :launch-specification     {:image-id 	       "ami-5cd51634"
                                        :instance-type 	       "t1.micro"
                                        :placement             {:availability-zone my-zone}
                                        :key-name	       "telekhine"
                                        :security-groups-ids   ["sg-78deaf1d"]
                                        :subnet-id             (get my-subnets my-zone)
                                        :iam-instance-profile  {:arn "arn:aws:iam::633840533036:instance-profile/girder-peer"}}})


)


#_(defn dns-names [is]
  (loop [n 5]
    (let [ds    (:reservations (describe-instances :instance-ids is))
          names (map #(get-in % [:instances 0 :public-dns-name]) ds)]
      (cond (or
           (and (seq names)
                (= (count names) (count is))
                (every? string? names))
           (zero? n))
        names
        (zero? n)
        nil
        :else
        (do
          (Thread/sleep 10000)
          (recur (dec n)))))))

#_(defn request-spots [n & {:keys [zone itype price image key profile group subnet]}]
  (let [zone   (or zone "us-east-1c")
        itype  (or itype "t1.micro")
        price  (str (or price "0.005"))
        image  (or image "BLEH")
        sgs    (if (seq sg) sg [sg])
        req    [:spot-price 			price
                :instance-count 		n
                :type 			"one-time"
                :launch-specification 
                {:image-id 			image
                 :instance-type 		itype
                 :placement {:availability-zone      zone}
                 :key-name			key
                 :security-groups-ids	     sgs
                 :subnet-id                  (get my-subnets zone)
                 :iam-instance-profile       {:arn "arn:aws:iam::633840533036:instance-profile/girder-peer"}}]
        (apply request-spot-instances req)]
    (map :spot-instance-request-id  (:spot-instance-requests r))))

#_(defn patience
  "Invoke (genfn) every time-wait sec until predicate returns true or time-limit sec has passed.
Returns result of genfn, or nil in the case of timeout."
  [pred genfn time-limit time-wait]
  (let [t1     (if (number? time-limit) (timeout (* 1000 time-limit)) time-limit)
        out    (chan)]
    (async/go-loop []
      (let [r   (genfn)
            p   (pred r)]
        (if p
         (do (debug "Returning" r)
              (>! out r) (close! out))
         (let [t2     (timeout (* 1000  time-wait))
               _      (debug "Still" r "Waiting" time-wait "sec")
               [v c] (async/alts! [t1 t2])]
           (if (= c t1)
             (do 
               (debug time-limit "seconds expired.  Returning nil.")
               (close! out))
             (recur))))))
    out))


#_(defn patience-every [pred collfn time-limit time-wait]
  (patience #(every? pred %) collfn time-limit time-wait))


#_(defn- add-sessions [res]
  (let [out   (chan)
        tout  (timeout 60000)
        rs    (:rs res)
        hosts (:hosts res)
        sess  (map #(ssh/session ag % {:strict-host-key-checking :no
                                                  :username "ec2-user"})
                   hosts)]
    (async/go-loop []
      (if (every? ssh/connected? sess)
        (do (debug "All connected.  Returning.")
            (>! out (assoc res :sessions sess)))
        (let 
          (if (= c tout)
            (do
              (debug "Failed to establish connections after timeout.  Terminating.")
              (cancel rs)
              (close! out))
            (if #(every? ssh/connected? sess)
              (do (debug "Returning connected" sess)
                  (>! out (assoc res :sessions sess)))
              (do
                (debug "Waiting 10 sec")
                (<! (timeout 10000))
                (recur)))))))
    out))


#_(defn- add-requests [res]
  (let [req (:req res)
        out (chan)]
    (go 
      (let [rs  (request-spots req)]
        (debug "waiting for" rs "to be active")
        (<! (timeout 1000))
        (if (<! (patience-every #(= "active" %) #(request-status rs) 600 60))
          (do (debug "All instances active! Returning" (assoc res :rs rs))
              (>! out (assoc res :rs rs)))
          (give-up-and-cancel res out "Requests failed.")
)))
    out))


#_(defn request-info [ids]
  (let [d      (cmd-timeout 5 describe-spot-instance-requests :spot-instance-request-ids (filter #(re-find #"^sir-" %) ids))
        sirs   (:spot-instance-requests d)
        rs     (map :spot-instance-request-id sirs)
        rs->rs (into {} (map #(vector %1 %2) rs0 rs))
        ss     (map :state sirs)
        is     (map :instance-id sirs)
        rs->i  (into {} (map #(vector %1 %2) rs is))
        is-up  (filter identity is)
        is-r   (:reservations (describe-instances :instance-ids is-up)) 
        is-i   (map #(get-in % [:instances 0]) is-r)
        ;ip-pub (map :public-ip-address is-i)
        is-up  (map :instance-id is-i)
        ip-pri (map :private-ip-address is-i)
        hosts  (map :public-dns-name is-i)
        hosts  (map #(when (seq %) %) hosts)
        i->inf (into {} (map #(vector %1 {:instance %1 :ip %2 :host %3}) is-up ip-pri hosts))
        ]
    (map #(merge {:request %1 :status %2} ((comp i->inf rs->i) %1)) rs0 ss)))


#_(defn patience
  "Invoke (genfn) every time-wait sec until predicate returns true or time-limit sec has passed.
Returns result of genfn, or nil in the case of timeout."
  [genfn pred failure time-limit time-wait]
  (let [t1     (timeout (* 1000 time-limit))
        out    (chan)]
    (async/go-loop []
      (let [rc     (go-try (genfn))
            [v c]  (async/alts! [rc t1 out])]
        (debug "Ats:" v c)
        (cond
         (= c out)                    (do (debug "Canceled")
                                          (when failure (failure)))
         (= c t1)                     (do (debug time-limit "sec expired")
                                          (when failure (failure))
                                          (close! out))
         (and (= c rc) v (pred v))    (do (debug "Returning" v)
                                          (>! out v))
         :else                        (do (debug "Still" v "Waiting" time-wait)
                                          (<! (timeout (* 1000 time-wait)))
                                          (debug "Woke up after wait...")
                                          (recur)))))
    out))


#_(defn pluck [c]
  (let [a (atom nil)]
    (close! (go (let [x (<! c)]
                  (reset! a x))))
    a))


#_(defn- add-sessions [req]
  (let [sess (map #(ssh/session ag % {:strict-host-key-checking :no
                                      :username "ec2-user"})
                  (:hosts req))]
    (patience
     (fn [] (let [sbad (filter #(not (ssh/connected? %)) sess)]
              (doseq [s sbad] (ssh/connect s))
              (assoc req :sessions sess)))
     #(every? ssh/connected? (:sessions %1))
     nil ;#(give-up-and-cancel req "Failed to add sesions")
     60 10
     )))

#_(defn chain-some
  "Each cfn must be a function of one parameter, returning a channel that will deliver
its non-nil result, or nil on failure.  Non-nil return values will be progressively passed
to each cfn in order.  Returns a channel that will contain the result of the final
cfn or nil, if one of them failed along the way."
  [x & cfns]
  (let [out (chan)]
    (debug "Starting" cfns)
    (async/go-loop [[f & fs] cfns
                    x        x]
      (if-not f
        (do (debug "Finished" cfns)
            (>! out x))
        (let [_ (debug "Invoking" f x)
              y (<! (f x))]
          (if-not y
            (do (debug "Failed at" f)
                (close! out))
            (recur fs y)))))
    out))


;; 
#_(defn bring-up-aws
  "Bring up n AWS t1.micro instances and return channel that will contain a single map
of {:instance ids :hosts names and :sessions objects}, with appropriate timeouts.
Returns a channel that will contain a map of :hosts, :sessions, etc or nil."
  [reqmap nmin & opts]
  (let [req (apply ph-assoc reqmap paths opts)
        req (assoc req :nmin nmin)
        n    (ph-get reqmap paths :n)
        req (if (<= nmin n) req (ph-assoc req paths :n nmin))]
    (debug "Bringing up" req)
    (chain-some {:req req} add-requests; add-instances add-ips add-hosts
                )
    ))
;; add-sessions



#_(defn- add-requests [req]
  (let [req (assoc req :rs (request-spots (:req req)))]
    (patience
     #(assoc req :req-status (cmd-timeout 5 request-status (:rs req)))
     (fn [req] (let [rss (:req-status req)]
                 (and (seq rss) (every? #(= "active" %) rss))))
   #(give-up-and-cancel req "Failed to fulfill spot requests")
   600 60)))

#_(defn- add-instances [req]
  (patience 
   #(assoc req :is (request-instances (:rs req)))
   #(good-strings (:is %))
   #(give-up-and-cancel req "Failed to get instance ids")
   10 1))


#_(defn- add-hosts [req]
  (patience
   (fn [] (assoc req :hosts (->> (:is req)
                                 (describe-instances :instance-ids)
                                 :reservations
                                 (map #(get-in % [:instances 0 :public-dns-name])))))
   #(good-strings (:hosts %))
   #(give-up-and-cancel req "Failed to get hosts.")
   10 1))


#_(defn- add-ips [req]
  (patience
   (fn [] (assoc req :ips (->> (:is req)
                                 (describe-instances :instance-ids)
                                 :reservations
                                 (map #(get-in % [:instances 0 :private-ip-address])))))
   #(good-strings (:ips %))
   #(give-up-and-cancel req "Failed to get IPs")
   10 1))

#_(defn- give-up-and-cancel [req & [msg]]
  (when msg (debug msg))
  (concat  (some-> req :is terminate)
           (some-> req :rs cancel)))


    (async/go-loop [i 0]
      (if (>= i nmin)
        (let [rs->info (request-info-map rs)
              pred     (partial really-up? rs->info)
              rs-up    (filter pred rs)
              rs-dn    (filter (complement pred) rs)]
          (close-notify-chan! id)
          (cancel-requests rs-dn)
          (map rs->info rs-up))
