<commit_before>(ns mail-sieve-e.core
  (:gen-class)
  (:require [clojure.core.async :as a :refer [go-loop chan <!! >!!
                                        go alt!! alt! timeout
                                        <! >!]]
            [mail-sieve-e.sieve :as s]
            [clojure.java.io :as io])
  (:import [java.net Socket ServerSocket]
           [java.io PrintWriter InputStreamReader BufferedReader]))

(declare read-handler write-handler)

(defn connect
  "Takes: server - host ip to connect to
          port   - port to connect to"
  [server port]
  (let [socket (Socket. server port)
        in (-> socket
               .getInputStream
               InputStreamReader.
               BufferedReader.
               (read-handler))
        out (-> socket
                .getOutputStream
                PrintWriter.
                (write-handler))]
    {:socket socket
     :in in
     :out out}))


(defn receive
  "Read a line of textual data from the given socket"
  [socket]
  (.readLine (io/reader socket)))

(defn write
  "Send the given string message out over the given socket"
  [socket msg]
  (let [writer (io/writer socket)]
    (.write writer (str msg "\n"))
    (.flush writer)))

(defn send-to-all
  "Sends a msg to all connected machines"
  [connected msg]
  (doall
   (map #(write % msg) @connected)))

(defn read-handler
  "Given a socket and returns a channel which is populated
  with all of information coming from that socket"
  [socket]
  (let [read-chan (chan 1e18)
        reader (io/reader socket)]
    (go-loop []
      (try
        (let [msg-in (.readLine reader)]
          (when-not (empty? msg-in)
            (>!! read-chan (read-string msg-in))))
        (catch Exception e (str "Socket closed.")))
      (recur))
    read-chan))

(defn write-handler
  "Given a socket it returns a channel which writes to the socket
  whatever is put into the channel"
  [socket]
  (let [write-chan (chan 1e18)
        writer (io/writer socket)]
    (go-loop []
      (when-let [msg (<! write-chan)]
        (when-not (nil? msg)
          (do
            (.write writer (str msg "\n"))
            (.flush writer)))
        (recur)))
    write-chan))

(defn async-echo-server
  "Takes: port      - port number
          handler   - a atom of a function that takes msgs from connected clients
          send-chan - A channel to immediately send information
  and starts an ayncronous server which handles all connections"
  [port handler send-chan]
  (let [running?      (atom true)
        connected     (atom [])
        server-socket (ServerSocket. port)]
    (try
      (go
        (while @running?
          (when-let [sock (.accept server-socket)]
            (swap! connected conj sock)
            (let [client-in (read-handler sock)]
              (go-loop []
                (when @running?
                  (do
                    (alt! [client-in] ([msg] (write sock (@handler msg)))
                          [send-chan] ([msg] (send-to-all connected msg)))
                    (recur))))))))
      (catch Exception e nil))
    (go-loop []
      (if-not running?
        (.close server-socket)
        (recur)))
    {:server-socket server-socket
     :running running?
     :connected connected}))

(defn wait-for-clients
  "Given a num-expected number of machines to connect
  and the list of currently connected computers waits until
  the expected number of computers connect."
  [num-expected connected]
  ; First, wait for all computers to join
  (println "Waiting for computers to join...")
  (while (not= (count @connected) num-expected)
    (Thread/sleep 2000))
    (println "# Connected: " (count @connected))
  (Thread/sleep 200))

(defn transfer-primes
  "All machines send lead their primes vector,
  and lead must send these to the right computers."
  [connections done? prime-vec]
  (let [[mi ps p] prime-vec
        r-comps (drop (dec mi) connections) ; computers we actually have to send the nums to
        appoint? (= ps -1)]
    ; Send prime-vec
    (mapv #(write % prime-vec) r-comps)
    ; Check if it's an appoint signal
    ; When ps = -1, appoint the next machine,
    ; since machines are indexed starting at 1, and vectors at 0,
    ; to get the next comp get comps at my-num
    ; ex machine 1 appoints machine two, [m1 m2 m3...], m2 at index 1
    (when (and appoint?
               (= (count connections) (dec mi)))
      (reset! done? true))))

(defn lead-start
  "Given: num-expected - number of machines to run on the network
          num-primes   - number to find all primes underneath
          port         - port to listen on"
  [num-expected num-primes port]
  (let [send-chan  (chan 10000000)
        handler    (atom #(.toUpperCase %))
        server     (async-echo-server port handler send-chan)
        connected  (:connected server)
        ; Wait for connected clients
        _          (wait-for-clients (dec num-expected) connected)
        connected  @connected ; Ensure connected computers stays constant
        done?      (atom false) ; Tells when we're finished
        ; Change handler function to transfer-primes function
        ___        (reset! handler (partial transfer-primes connected done?))
        chunks     (s/spread-work num-primes num-expected)
        lead-chunk (s/gen-table (first chunks))]
    ; First thing to do is send machine numbers and bounds
    (doall
     (for [mi (range (count connected))
           :let [machine (get connected mi)
                 bounds  (get chunks (inc mi))]] ; first chunk is for lead
       (do
         (write machine (+ mi 2))
         (write machine bounds))))
    ; Now we can start the sieve
    (s/sieve-e 1 true (chan 10) lead-chunk send-chan)
    (println "Waiting for other machines to finish...\n")
    (while (not @done?)
      (<!! (timeout 500)))
    ; shut down server
    (println "Shutting down server...\n")
    (do
      (>!! send-chan 0)
      (<!! (timeout 200))
      (reset! (:running server) false)
      (mapv #(.close %) connected)
      (a/close! send-chan)
      (.close (:server-socket server)))
    (println "Sieve completed!")))

(defn client-start
  "Starts a following process listening on host:port"
  [host port]
  (println "connecting to host...")
  (let [lead (connect host port)]
    (println "connected to host!\n")
    (println "waiting for all other computers to connect...\n")
    (let [my-num (int(<!! (:in lead)))
          bounds (mapv int (<!! (:in lead)))
          chunk  (s/gen-table bounds)]
      (when-let [start? (<!! (:in lead))]
        (s/sieve-e my-num false (:in lead) chunk (:out lead))
        (println "Waiting for kill signal...")
        (while (not= 0 (<!! (:in lead)))
          (<!! (timeout 50)))
        (do (a/close! (:in lead))
            (a/close! (:out lead)))
        (println "Done!")))))

(defn -main
  "Distributed implementation of the Sieve of Eratosthenes"
  ([num-comps num-primes port]
   (lead-start (Integer. num-comps) (Integer. num-primes) (Integer. port)))
  ([host port]
   (client-start host (Integer. port))))<commit_msg>added documentation and fixed some bugs<commit_after>(ns mail-sieve-e.core
  (:gen-class)
  (:require [clojure.core.async :as a :refer [go-loop chan <!! >!!
                                        go alt!! alt! timeout
                                        <! >!]]
            [mail-sieve-e.sieve :as s]
            [clojure.java.io :as io])
  (:import [java.net Socket ServerSocket]
           [java.io PrintWriter InputStreamReader BufferedReader]))

(declare read-handler write-handler)

(defn connect
  "Takes: server - host ip to connect to
          port   - port to connect to
  and returns a map of channels to read/write to
  and a raw socket"
  [server port]
  (let [socket (Socket. server port)
        in (-> socket
               .getInputStream
               InputStreamReader.
               BufferedReader.
               (read-handler))
        out (-> socket
                .getOutputStream
                PrintWriter.
                (write-handler))]
    {:socket socket
     :in in
     :out out}))


(defn receive
  "Read a line of textual data from the given socket"
  [socket]
  (.readLine (io/reader socket)))

(defn write
  "Send the given string message out over the given socket"
  [socket msg]
  (let [writer (io/writer socket)]
    (.write writer (str msg "\n"))
    (.flush writer)))

(defn read-handler
  "Given a socket and returns a channel which is populated
  with all of information coming from that socket"
  [socket]
  (let [read-chan (chan 1e18)
        reader (io/reader socket)]
    (go-loop []
      (try
        (let [msg-in (.readLine reader)]
          (when-not (empty? msg-in)
            (>!! read-chan (read-string msg-in))))
        (catch Exception e (str "Socket closed.")))
      (recur))
    read-chan))

(defn write-handler
  "Given a socket it returns a channel which writes to the socket
  whatever is put into the channel"
  [socket]
  (let [write-chan (chan 1e18)
        writer (io/writer socket)]
    (go-loop []
      (when-let [msg (<! write-chan)]
        (when-not (nil? msg)
          (do
            (.write writer (str msg "\n"))
            (.flush writer)))
        (recur)))
    write-chan))

(defn async-echo-server
  "Takes: port      - port number
          handler   - a atom of a function that takes msgs from connected clients
          send-chan - A channel to immediately send information
  and starts an ayncronous server which handles all connections"
  [port handler send-chan]
  (let [running?      (atom true)
        connected     (atom [])
        server-socket (ServerSocket. port)]
    (try
      (go
        (while @running?
          (when-let [sock (.accept server-socket)]
            (swap! connected conj sock)
            (let [client-in (read-handler sock)]
              (go-loop []
                (when @running?
                  (do
                    (alt! [client-in] ([msg] (write sock (@handler msg)))
                          [send-chan] ([msg] (doall (map #(write % msg) @connected))))
                    (recur))))))))
      (catch Exception e nil))
    (go-loop []
      (if-not running?
        (.close server-socket)
        (recur)))
    {:server-socket server-socket
     :running running?
     :connected connected}))

(defn wait-for-clients
  "Given a num-expected number of machines to connect
  and the list of currently connected computers waits until
  the expected number of computers connect."
  [num-expected connected]
  ; First, wait for all computers to join
  (println "Waiting for computers to join...")
  (while (not= (count @connected) num-expected)
    (Thread/sleep 2000))
    (println "# Connected: " (count @connected))
  (Thread/sleep 200))

(defn transfer-primes
  "All machines send lead their primes vector,
  and lead must send these to the right computers."
  [connections done? prime-vec]
  (let [[mi ps p] prime-vec
        r-comps (drop (dec mi) connections) ; computers we actually have to send the nums to
        appoint? (= ps -1)]
    ; Send prime-vec
    (mapv #(write % prime-vec) r-comps)
    ; Check if it's an appoint signal
    ; When ps = -1, appoint the next machine,
    ; since machines are indexed starting at 1, and vectors at 0,
    ; to get the next comp get comps at my-num
    ; ex machine 1 appoints machine two, [m1 m2 m3...], m2 at index 1
    (when (and appoint?
               (= (count connections) (dec mi)))
      (reset! done? true))))

(defn lead-start
  "Given: num-expected - number of machines to run on the network
          num-primes   - number to find all primes underneath
          port         - port to listen on"
  [num-expected num-primes port]
  (let [send-chan  (chan 10000000)
        handler    (atom #(.toUpperCase %))
        server     (async-echo-server port handler send-chan)
        connected  (:connected server)
        ; Wait for connected clients
        _          (wait-for-clients (dec num-expected) connected)
        connected  @connected ; Ensure connected computers stays constant
        done?      (atom false) ; Tells when we're finished
        ; Change handler function to transfer-primes function
        ___        (reset! handler (partial transfer-primes connected done?))
        chunks     (s/spread-work num-primes num-expected)
        lead-chunk (s/gen-table (first chunks))]
    ; First thing to do is send machine numbers and bounds
    (doall
     (for [mi (range (count connected))
           :let [machine (get connected mi)
                 bounds  (get chunks (inc mi))]] ; first chunk is for lead
       (do
         (write machine (+ mi 2))
         (write machine bounds)
         (write machine 1))))
    ; Now we can start the sieve
    (s/sieve-e 1 true (chan 10) lead-chunk send-chan)
    (println "Waiting for other machines to finish...\n")
    (while (not @done?)
      (<!! (timeout 500)))
    ; shut down server
    (println "Shutting down server...\n")
    (do
      ; send kill signal
      (>!! send-chan 0)
      ; wait for machines to recieve signal
      (<!! (timeout 200))
      ; shutdown various parts
      (reset! (:running server) false)
      (mapv #(.close %) connected)
      (a/close! send-chan)
      (.close (:server-socket server)))
    (println "Sieve completed!")))

(defn client-start
  "Starts a following process listening on host:port"
  [host port]
  (println "connecting to host...")
  ; first connect to lead computer
  (let [lead (connect host port)]
    (println "connected to host!\n")
    (println "waiting for all other computers to connect...\n")
    ; recieve machine number and bounds while conveniently waiting for lead to send info
    (let [my-num (int(<!! (:in lead)))
          bounds (mapv int (<!! (:in lead)))
          chunk  (s/gen-table bounds)]
      ; wait for start signal
      (when-let [start? (<!! (:in lead))]
        ; run sieve listening on lead channels
        (s/sieve-e my-num false (:in lead) chunk (:out lead))
        ; then wait for kill signal
        (println "Waiting for kill signal...")
        (while (not= 0 (<!! (:in lead)))
          (<!! (timeout 50)))
        ; Then close the channels and socket
        (do (a/close! (:in lead))
            (a/close! (:out lead))
            (.close (:socket lead)))
        (println "Done!")))))

(defn -main
  "Distributed implementation of the Sieve of Eratosthenes"
  ([num-comps num-primes port]
   (lead-start (Integer. num-comps) (Integer. num-primes) (Integer. port)))
  ([host port]
   (client-start host (Integer. port))))
{'fraction_non_alphanumeric': 0.11948908117016893, 'fraction_numerical': 0.006249141601428376, 'mean_word_length': 2.12913622690159, 'pattern_counts': {'":': 0, '<': 23, '<?xml version=': 0, '>': 15, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 2, 'word_list_counts': {'cursed_substrings.json': 8, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7288377', 'n_tokens_mistral': 4387, 'n_tokens_neox': 4173, 'n_words': 1560}