Có một thành ngữ Clojure để gửi nhiều biểu thức song song

Question

Tôi có một số biểu thức (chưa được đánh giá) được tổ chức trong một vectơ; [expr1 expr2 expr3 ...]

Điều tôi muốn làm là đưa từng biểu thức cho một chuỗi riêng biệt và đợi cho đến khi một biểu thức trả về một giá trị. Tại thời điểm đó tôi không quan tâm đến kết quả từ các chủ đề khác và muốn hủy bỏ chúng để tiết kiệm tài nguyên CPU.

(Tôi nhận ra rằng điều này có thể gây ra không thuyết định mệnh ở chỗ chạy khác nhau của chương trình có thể gây ra biểu hiện khác nhau được đánh giá đầu tiên. Tôi có điều này trong tay.)

Có một tiêu chuẩn/cách thành ngữ đạt được ở trên?

Answer 1

Đây là sự lựa chọn của tôi.

Về cơ bản bạn cần phải giải quyết một lời hứa toàn cầu bên trong mỗi người tương lai của bạn, sau đó trả về một vector chứa danh sách tương lai và giá trị giải quyết và sau đó hủy bỏ tất cả các kỳ hạn trong danh sách:

(defn run-and-cancel [& expr] 
    (let [p (promise) 
      run-futures (fn [& expr] [(doall (map #(future (deliver p (eval %1))) expr)) @p]) 
      [fs res] (apply run-futures expr)] 
     (map future-cancel fs) 
     res)) 

Answer 2

Mặc dù tôi không biết liệu có một cách thành ngữ để đạt được mục tiêu của bạn hay không nhưng Clojure Future có vẻ phù hợp.

Mất một cơ thể của biểu thức và mang lại một đối tượng trong tương lai mà sẽ gọi cơ thể trong chủ đề khác, và sẽ cache kết quả và gửi lại trên tất cả các cuộc gọi tiếp theo để deref/@. Nếu tính toán có chưa hoàn thành, các cuộc gọi đến deref/@ sẽ chặn, trừ khi biến thể của deref với thời gian chờ được sử dụng.

Answer 3

Điều bạn muốn là số CompletionService của Java. Tôi không biết bất kỳ cái bao bọc nào quanh cái áo choàng này, nhưng nó sẽ không khó để làm với interop. Ví dụ bên dưới dựa trên ví dụ trên trang JavaDoc cho ExecutorCompletionService.

(defn f [col] 
    (let [cs (ExecutorCompletionService. (Executors/newCachedThreadPool)) 
      futures (map #(.submit cs %) col) 
      result (.get (.take cs))] 
     (map #(.cancel % true) futures) 
     result)) 

Answer 4

Nó không đạt được phát hành chính thức, nhưng core.async vẻ như nó có thể là một cách thú vị giải quyết vấn đề của bạn - và vấn đề đồng bộ khác, rất gọn gàng.

Các bùa leiningen cho core.async là (hiện tại) như sau:

[org.clojure/core.async "0.1.0-SNAPSHOT"] 

Và đây là một số mã để thực hiện một chức năng đó sẽ mất một số chức năng tốn nhiều thời gian, và ngăn chặn cho đến khi một trong số họ trở về.

(require '[clojure.core.async :refer [>!! chan alts!! thread]]))  

(defn return-first [& ops] 
    (let [v (map vector ops (repeatedly chan))] 
    (doseq [[op c] v] 
     (thread (>!! c (op)))) 
    (let [[value channel] (alts!! (map second v))] 
     value))) 

;; Make sure the function returns what we expect with a simple Thread/sleep 
(assert (= (return-first (fn [] (Thread/sleep 3000) 3000) 
         (fn [] (Thread/sleep 2000) 2000) 
         (fn [] (Thread/sleep 5000) 5000)) 
      2000)) 

Trong ví dụ trên:

• chan tạo ra một kênh không đồng bộ
• >!! đặt một giá trị vào một kênh
• thread thực hiện cơ thể trong một thread
• alts!! mất một vector kênh và trả về khi giá trị xuất hiện trên bất kỳ kênh nào trong số đó

Có cách hơn để nó hơn này, và tôi vẫn nhận được vòng đầu của tôi nó, nhưng có một hương ở đây: https://github.com/clojure/core.async/blob/master/examples/walkthrough.clj

Và blog của David Nolen của có một số rất lớn, nếu tâm-boggling, bài viết trên nó (http://swannodette.github.io/)

Sửa

Chỉ cần thấy rằng Michał Marczyk đã trả lời một câu hỏi rất giống nhau, nhưng tốt hơn, ở đây, và nó cho phép bạn hủy bỏ/ngắn mạch. with Clojure threading long running processes and comparing their returns

Answer 5

Bạn có thể sử dụng cuộc gọi trong tương lai để nhận danh sách tất cả tương lai, lưu trữ chúng trong Atom. sau đó, soạn mỗi tương lai đang chạy với chức năng "bắn những cái khác trong đầu" để cái đầu tiên sẽ kết thúc tất cả những cái còn lại. Here is an example:

(defn first-out [& fns] 
    (let [fs (atom []) 
     terminate (fn [] (println "cancling..") (doall (map future-cancel @fs)))] 
    (reset! fs (doall (map (fn [x] (future-call #((x) (terminate)))) fns))))) 

(defn wait-for [n s] 
    (fn [] (print "start...") (flush) (Thread/sleep n) (print s) (flush))) 

(first-out (wait-for 1000 "long") (wait-for 500 "short")) 

Sửa

Chỉ cần nhận thấy rằng mã trước đó không trả lại kết quả đầu tiên, vì vậy nó là chủ yếu hữu ích cho tác dụng phụ. ở đây là another version trả về kết quả đầu tiên khi sử dụng lời hứa:

(defn first-out [& fns] 
    (let [fs (atom []) 
     ret (promise) 
     terminate (fn [x] (println "cancling..") 
          (doall (map future-cancel @fs)) 
          (deliver ret x))] 
    (reset! fs (doall (map (fn [x] (future-call #(terminate (x)))) fns))) 
    @ret)) 

(defn wait-for [n s] 
    "this time, return the value" 
    (fn [] (print "start...") (flush) (Thread/sleep n) (print s) (flush) s)) 

(first-out (wait-for 1000 "long") (wait-for 500 "short"))