<commit_before>(ns morpheus.traversal.bfs
  (:require [neb.core :as neb]
            [morpheus.computation.base :as compute]
            [morpheus.messaging.core :as msg]
            [morpheus.models.vertex.core :as vertex]
            [neb.utils :refer :all]
            [neb.base :as nb]
            [cluster-connector.distributed-store.core :as ds]
            [com.climate.claypoole :as cp]
            [morpheus.models.edge.core :as edges]
            [morpheus.models.edge.base :as eb]
            [morpheus.query.lang.evaluation :as eva]
            [morpheus.traversal.bfs.rebuild :refer :all]
            [clojure.core.async :as a]
            [cluster-connector.utils.for-debug :refer [$ spy]]))

; Parallel breadth-first-search divised by <NAME>
; The algoithm was first introduced in Lawrence National Laboratory on BlueGene supercomputer
; I don't really sure my impelementation is 100% the same as theirs. But the ideas are similiar.
; I also took some idea from <NAME> and <NAME>, for the underlying BSP data processing mechanism.
; Morpueus will use the task founder server as the only machine for join operations in each level.

(def tasks-vertices (atom {}))
(def superstep-tasks (atom {}))

(defn partation-vertices [vertex-ids]
  (group-by
    #(nb/locate-cell-by-id %)
    vertex-ids))

(defn proc-forward-msg [task-id data]
  (let [[superstep-id vertices-stack] data
        {:keys [founder filters with-edges? with-vertices?] :as extra-params} (compute/get-task task-id)]
    (msg/send-msg
      founder :BFS-RETURN
      [superstep-id
       (->> (cp/pfor
              compute/compution-threadpool
              [vertex-id vertices-stack]
              (let [vertex (vertex/vertex-by-id vertex-id)
                    vertex-criteria (get-in filters [:criteria :vertex])
                    vertex-vailed (if vertex-criteria (eva/eval-with-data vertex vertex-criteria) true)]
                (when vertex-vailed
                  (let [neighbours (apply edges/neighbours-edges vertex (if filters (mapcat identity filters) []))
                        vertex-res (if with-vertices? vertex (select-keys vertex [:*id*]))
                        edges-res (map (fn [edge]
                                         (assoc (if with-edges? edge {})
                                           :*opp* (eb/get-oppisite edge vertex-id)))
                                       neighbours)]
                    [vertex-res edges-res]))))
            (filter identity)
            (doall))]
      :task-id task-id)))

(defn proc-return-msg [task-id data]
  (let [[superstep-id vertices-stack] data
        deepth (get-in @tasks-vertices [task-id :current-level])]
    (a/go
      (cp/pdoseq<commit_msg>possible lazy seq<commit_after>(ns morpheus.traversal.bfs
  (:require [neb.core :as neb]
            [morpheus.computation.base :as compute]
            [morpheus.messaging.core :as msg]
            [morpheus.models.vertex.core :as vertex]
            [neb.utils :refer :all]
            [neb.base :as nb]
            [cluster-connector.distributed-store.core :as ds]
            [com.climate.claypoole :as cp]
            [morpheus.models.edge.core :as edges]
            [morpheus.models.edge.base :as eb]
            [morpheus.query.lang.evaluation :as eva]
            [morpheus.traversal.bfs.rebuild :refer :all]
            [clojure.core.async :as a]
            [cluster-connector.utils.for-debug :refer [$ spy]]))

; Parallel breadth-first-search divised by <NAME>
; The algoithm was first introduced in Lawrence National Laboratory on BlueGene supercomputer
; I don't really sure my impelementation is 100% the same as theirs. But the ideas are similiar.
; I also took some idea from <NAME> and <NAME>, for the underlying BSP data processing mechanism.
; Morpueus will use the task founder server as the only machine for join operations in each level.

(def tasks-vertices (atom {}))
(def superstep-tasks (atom {}))

(defn partation-vertices [vertex-ids]
  (group-by
    #(nb/locate-cell-by-id %)
    vertex-ids))

(defn proc-forward-msg [task-id data]
  (let [[superstep-id vertices-stack] data
        {:keys [founder filters with-edges? with-vertices?] :as extra-params} (compute/get-task task-id)]
    (msg/send-msg
      founder :BFS-RETURN
      [superstep-id
       (->> (cp/pfor
              compute/compution-threadpool
              [vertex-id vertices-stack]
              (let [vertex (vertex/vertex-by-id vertex-id)
                    vertex-criteria (get-in filters [:criteria :vertex])
                    vertex-vailed (if vertex-criteria (eva/eval-with-data vertex vertex-criteria) true)]
                (when vertex-vailed
                  (let [neighbours (apply edges/neighbours-edges vertex (if filters (mapcat identity filters) []))
                        vertex-res (if with-vertices? vertex (select-keys vertex [:*id*]))
                        edges-res (doall (map (fn [edge]
                                                (assoc (if with-edges? edge {})
                                                  :*opp* (eb/get-oppisite edge vertex-id)))
                                              neighbours))]
                    [vertex-res edges-res]))))
            (filter identity)
            (doall))]
      :task-id task-id)))

(defn proc-return-msg [task-id data]
  (let [[superstep-id vertices-stack] data
        deepth (get-in @tasks-vertices [task-id :current-level])]
    (a/go
      (cp/pdoseq
{'fraction_non_alphanumeric': 0.11613848690236307, 'fraction_numerical': 0.001099102399706906, 'mean_word_length': 1.9787234042553192, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 13, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 11, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1766756', 'n_tokens_mistral': 1670, 'n_tokens_neox': 1594, 'n_words': 440}