<commit_before>(ns elephantdb.service
  (:import [java.util.concurrent.locks ReentrantReadWriteLock])
  (:import [elephantdb.generated ElephantDB ElephantDB$Iface])
  (:import [elephantdb Shutdownable client])
  (:import [elephantdb.persistence LocalPersistence])
  (:import [elephantdb.store DomainStore])
  (:require [elephantdb [domain :as domain] [thrift :as thrift] [shard :as shard]])
  (:use [elephantdb config log util hadoop loader]))

; { :replication 2
;   :hosts ["elephant1.server" "elephant2.server" "elephant3.server"]
;   :port 3578
;   :domains {"graph" "s3n://mybucket/elephantdb/graph"
;             "docs"  "/data/docdb"
;             }
; }

(defn- init-domain-info-map [fs global-config]
  (let [domain-shards (shard/shard-domains fs global-config)]
    (into {}
      (dofor [domain (keys (:domains global-config))]
          [domain (domain/init-domain-info (domain-shards domain))]
            ))))

(defn load-and-sync-status [domains-info loader-fn]
  (let [loaders (dofor [domain (keys domains-info)]
                       (future
                         (try
                           (let [domain-data (loader-fn domain)]
                             (if domain-data
                               (domain/set-domain-data! (domains-info domain)
                                                        domain-data))
                             (domain/set-domain-status!
                              (domains-info domain)
                              (thrift/ready-status false)))
                           (catch Throwable t
                             (log-error t "Error when loading domain " domain)
                             (domain/set-domain-status!
                              (domains-info domain)
                              (thrift/failed-status t))
                             ))))]
    (with-ret (future-values loaders)
      (log-message "Successfully loaded all domains"))
    ))

(defn domain-needs-update? [local-vs remote-vs]
  (or (nil? (.mostRecentVersion local-vs))
      (< (.mostRecentVersion local-vs)
         (.mostRecentVersion remote-vs))))

(defn use-cache-or-update
  ([domain domains-info global-config local-config open-if-no-update]
     (let [fs (filesystem (:hdfs-conf local-config))
           lfs (local-filesystem)
           local-dir (:local-dir local-config)
           local-domain-root (str (path local-dir domain))
           remote-path (-> global-config :domains (get domain))
           remote-vs (DomainStore. fs remote-path)
           local-vs (DomainStore. lfs local-domain-root (.getSpec remote-vs))]
       (if (domain-needs-update? local-vs remote-vs)
         (load-domain fs
                      local-config
                      local-domain-root
                      remote-path
                      (domain/host-shards (domains-info domain)))
         ;; use cached domain from local-dir (no update needed)
         (do
           (swap! finished-loaders + 1)
           (if open-if-no-update
             (open-domain local-config<commit_msg>Refactoring: moved domain load & sync out to load-and-sync-status-of-domain function<commit_after>(ns elephantdb.service
  (:import [java.util.concurrent.locks ReentrantReadWriteLock])
  (:import [elephantdb.generated ElephantDB ElephantDB$Iface])
  (:import [elephantdb Shutdownable client])
  (:import [elephantdb.persistence LocalPersistence])
  (:import [elephantdb.store DomainStore])
  (:require [elephantdb [domain :as domain] [thrift :as thrift] [shard :as shard]])
  (:use [elephantdb config log util hadoop loader]))

; { :replication 2
;   :hosts ["elephant1.server" "elephant2.server" "elephant3.server"]
;   :port 3578
;   :domains {"graph" "s3n://mybucket/elephantdb/graph"
;             "docs"  "/data/docdb"
;             }
; }

(defn- init-domain-info-map [fs global-config]
  (let [domain-shards (shard/shard-domains fs global-config)]
    (into {}
      (dofor [domain (keys (:domains global-config))]
          [domain (domain/init-domain-info (domain-shards domain))]
          ))))

(defn load-and-sync-status-of-domain [domains-info domain loader-fn]
  (future
    (try
      (let [domain-data (loader-fn domain)]
        (if domain-data
          (domain/set-domain-data! (domains-info domain)
                                   domain-data))
        (domain/set-domain-status!
         (domains-info domain)
         (thrift/ready-status false)))
      (catch Throwable t
        (log-error t "Error when loading domain " domain)
        (domain/set-domain-status!
         (domains-info domain)
         (thrift/failed-status t))))))

(defn load-and-sync-status [domains-info loader-fn]
  (let [loaders (dofor [domain (keys domains-info)]
                       (load-and-sync-status-of-domain domains-info domain loader-fn))]
    (with-ret (future-values loaders)
      (log-message "Successfully loaded all domains"))
    ))

(defn domain-needs-update? [local-vs remote-vs]
  (or (nil? (.mostRecentVersion local-vs))
      (< (.mostRecentVersion local-vs)
         (.mostRecentVersion remote-vs))))

(defn use-cache-or-update
  ([domain domains-info global-config local-config open-if-no-update]
     (let [fs (filesystem (:hdfs-conf local-config))
           lfs (local-filesystem)
           local-dir (:local-dir local-config)
           local-domain-root (str (path local-dir domain))
           remote-path (-> global-config :domains (get domain))
           remote-vs (DomainStore. fs remote-path)
           local-vs (DomainStore. lfs local-domain-root (.getSpec remote-vs))]
       (if (domain-needs-update? local-vs remote-vs)
         (load-domain fs
                      local-config
                      local-domain-root
                      remote-path
                      (domain/host-shards (domains-info domain)))
         ;; use cached domain from local-dir (no update needed)
         (do
           (swap! finished-loaders + 1)
           (if open-if-no-update
             (open-domain local-config
{'fraction_non_alphanumeric': 0.1251249583472176, 'fraction_numerical': 0.003332222592469177, 'mean_word_length': 1.9955089820359282, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4015750', 'n_tokens_mistral': 1805, 'n_tokens_neox': 1721, 'n_words': 408}