distributed writes 1

Author: bilbof

.gitignore

@@ -11,4 +11,6 @@ wheels/
 
 # data
 trie_checkpoints
-trie_wal
+trie_wal
+
+tmp/

README.md

@@ -40,6 +40,17 @@ A node in the ring owns some portion of the trie. The size of these portions is
 
 Increasing k-factor decreases memory usage of tries (since trie can be split into smaller shards) at a cost of slightly increasing memory usage of coordinator (rangemap). Main issue is functional since you can't currently search for a trie with a length less than the k-factor.
 
+## Components
+
+* API
+* Coordinator
+* ShardedTrie
+* LocalDB
+* Gossip
+* ~Registration
+
+Request flow: API -> Coordinator -> ShardedTrie -> LocalDB
+
 ## Developing locally
 
 ```sh

checkpoint.py

@@ -6,9 +6,9 @@
 import glob
 
 class Checkpoint:
-  def __init__(self, config, distributed_trie):
+  def __init__(self, config, sharded_trie):
     self.config = config
-    self.distributed_trie = distributed_trie
+    self.sharded_trie = sharded_trie
     self.dir = config.current()["checkpoint_dir"]
     self._ensure_dir()
 
@@ -20,7 +20,7 @@ def load(self):
       for line in f.readlines():
         prefix, object = line.rstrip().split(" ", maxsplit=1)
         trie_dict = json.loads(object)
-        self.distributed_trie.load(prefix, trie_dict)
+        self.sharded_trie.load(prefix, trie_dict)
     return True
 
 
@@ -29,7 +29,7 @@ def save(self):
     timestamp = str(int(time.time() * 100))
     tmp = open(os.path.join(os.getcwd(), self.dir, timestamp), "a+")
     try:
-      for prefix, trie in self.distributed_trie.range_tries.items():
+      for prefix, trie in self.sharded_trie.range_tries.items():
         print(f"{prefix} {json.dumps(trie.dict())}", file=tmp)
       checkpointed = True
     finally:
@@ -68,13 +68,12 @@ def _ensure_dir(self):
 
 
 # from config import Config
-# from distributed_trie import DistributedTrie
+# from sharded_trie import ShardedTrie
 # from wal import Wal
 
 # config = Config("config.yaml")
 # wal = Wal(config)
-# distributed_trie = DistributedTrie(config, wal)
-# trie = distributed_trie
+# trie = ShardedTrie(config, wal)
 
 # trie.add("twitter")
 # trie.add("twitch")

coordinator.py

@@ -2,23 +2,45 @@
 import random
 
 class Coordinator:
-  def __init__(self, config, cluster, rangemap):
+  def __init__(self, config, cluster, rangemap, local_trie):
     self.config = config
     self.cluster = cluster
     self.rangemap = rangemap
+    self.local_trie = local_trie
 
-  def matches(self, query, nodes):
-    node = self.random_healthy_node(nodes)
+  def matches(self, query):
+    preference_list = self.preference_list(query)
+    if self.config.node_id in preference_list:
+      return self.local_trie.matches(query)
+    node = self.random_healthy_node(preference_list)
     if not node: return []
     try:
       req = httpx.get(f"{node}/search", params={"q": query})
       if req.status_code == 200:
-        # todo decode
         return req.json()
     except Exception as err:
       print(f"Unexpected {err=}, when proxying to {node} with query {query} {type(err)=}")
       return []
 
+  def add(self, word):
+    return self._write_word("POST", word)
+  
+  def remove(self, word):
+    return self._write_word("DELETE", word)
+  
+  def preference_list(self, word):
+    k_factor = self.config.current()["k_factor"]
+    prefix, rem = word[:k_factor], word[k_factor+1:]
+    if len(prefix) != k_factor:
+      # We don't support word lengths at or below the k-factor
+      return []
+    return self.healthy_nodes(self.rangemap.nodes_for_key(prefix))
+
+  def healthy_nodes(self, nodes):
+    if len(nodes) == 0: return set()
+    pnodes = self.cluster.nodes
+    return set([node["host"] for node in pnodes if node["host"] in nodes and node["status"]["healthy"]])
+
   def random_healthy_node(self, node_hosts):
     nodes = []
     # todo change cluster.nodes to hash
@@ -27,17 +49,39 @@ def random_healthy_node(self, node_hosts):
     if len(nodes) == 0: return
     return random.choice(nodes)
 
-  def add(self, word, nodes):
-    return self._write_word("POST", nodes, word)
-  
-  def remove(self, word, nodes):
-    return self._write_word("DELETE", nodes, word)
-  
-  def _write_word(self, method, nodes, word):
-    # TODO: proxy in parallel to <replication_factor> healthy nodes, not just one
-    node = self.random_healthy_node(nodes)
-    if not node: return False
-    return self._proxy_request(method, f"{node}/api/v1/words", { "word": word })
+  def _write_word(self, method, word):
+    # TODO: keep tombstone (and check at coordinator if in local tombstone before sending unnecessary reqs)
+    # TODO: this whole thing is gross
+    preference_list = self.preference_list(word.word)
+    if len(preference_list) == 0: return False
+    no_failures = True
+    quorum = int((self.config.replication_factor / 2) + 1) # todo: allow configuring quorum separately
+    is_member = self.config.node_id in preference_list
+    if self.config.node_id in preference_list:
+      quorum -= 1
+      no_failures = no_failures and self.local_write(method, word.word)
+    
+    if quorum > 0 and no_failures and word.gossip:
+      visited = set()
+      if is_member:
+        visited.add(self.config.node_id)
+      # TODO: send in parallel, handle failures (try another node if one fails, sloppy quorum etc)
+      # TODO: rollbacks
+      for node in preference_list:
+        if node in visited: continue
+        if quorum <= 0: return True
+        quorum -= 1
+        self._proxy_request(method, f"{node}/api/v1/words", { "word": word.word, "gossip": False })  
+    return quorum <= 0 and no_failures
+    
+  def local_write(self, method, word):
+    match method:
+      case "POST":
+        return self.local_trie.add(word)
+      case "DELETE":
+        return self.local_trie.remove(word)
+      case _:
+        return False
 
   def _proxy_request(self, method, path, word, data=None):
     try:

main.py

@@ -1,5 +1,5 @@
 from typing import Union
-from pydantic import BaseModel
+from pydantic import BaseModel, Field
 from fastapi import FastAPI, APIRouter
 from config import Config
 from cluster import Cluster
@@ -9,7 +9,7 @@
 from apscheduler.schedulers.background import BackgroundScheduler
 from apscheduler.triggers.interval import IntervalTrigger
 import base64
-from distributed_trie import DistributedTrie
+from sharded_trie import ShardedTrie
 from wal import Wal
 from checkpoint import Checkpoint
 from coordinator import Coordinator
@@ -18,19 +18,11 @@
 cluster = Cluster(config)
 rangemap = RangeMap(cluster, config)
 wal = Wal(config)
-coordinator = Coordinator(config, cluster, rangemap)
-distributed_trie = DistributedTrie(config, wal, rangemap, coordinator)
-trie = distributed_trie
+trie = ShardedTrie(config, wal) # TODO: rename to trieDB
+coordinator = Coordinator(config, cluster, rangemap, trie)
 checkpointer = Checkpoint(config, trie)
-if not checkpointer.load() and distributed_trie.replay_wal():
-  trie.add("twitter")
-  trie.add("twitch")
-  trie.add("twilight")
-  trie.add("twigs")
-  trie.add("twig")
-  trie.add("tough")
-  trie.add("thought")
-
+checkpointer.load()
+trie.replay_wal()
 
 def node_healthchecks():
   print(f"Checking health of nodes at {datetime.now()}")
@@ -87,7 +79,7 @@ def prefix_to_nodes(q: str):
 
 @app.get("/search")
 def search(q: str):
-  return distributed_trie.matches(q)
+  return coordinator.matches(q)
 
 # API
 
@@ -99,17 +91,25 @@ def search(q: str):
 
 class Word(BaseModel):
   word: str
+  gossip: bool = Field(default=True)
 
 @api_router.post("/words")
 async def create_word(word: Word):
   # check added strings are in lexicon
-  trie.add(word.word)  
-  return { "message": f"Added '{word.word}' to Trifecta db" }
+  ok = coordinator.add(word)
+  if ok:
+    return { "message": f"Added '{word.word}' to Trifecta db" }
+  else:
+    return { "error": f"Did not add '{word.word}' to Trifecta db" }
 
 @api_router.delete("/words")
 async def delete_word(word: Word):
-  trie.remove(word.word)
-  return { "message": f"Deleted '{word.word}' from Trifecta db" }
+  ok = coordinator.remove(word)
+  if ok:
+    return { "message": f"Removed '{word.word}' to Trifecta db" }
+  else:
+    return { "error": f"Did not remove '{word.word}' to Trifecta db" }
+
 
 app.include_router(api_router)
 

distributed_trie.py sharded_trie.py

@@ -5,27 +5,19 @@
 from datetime import datetime
 import threading
 
-class DistributedTrie:
-  def __init__(self, config, wal, rangemap, coordinator):
+# TODO: Extract KV store from this, provide it as API for in-mem component
+class ShardedTrie:
+  def __init__(self, config, wal):
     self.range_tries = {}
     self.config = config
     self.checkpoint_dir = config.current()["checkpoint_dir"]
     self.wal = wal
-    self.rangemap = rangemap
-    self.coordinator = coordinator
     # TODO: Lock per trie
     self.sem = threading.Semaphore()
 
   def matches(self, query):
     k_factor = self.config.current()["k_factor"]
     prefix, rem = query[:k_factor], query[k_factor+1:]
-    if len(prefix) != k_factor:
-      # We don't support prefix search at or below the k-factor
-      return []
-    preference_list = self.rangemap.nodes_for_key(prefix)
-    if self.config.node_id not in preference_list:
-      print("Forwarding")
-      return self.coordinator.matches(query, preference_list)
     if prefix not in self.range_tries:
       return []
     return self.range_tries[prefix].matches(query)
@@ -37,15 +29,8 @@ def add(self, word, commit_to_wal=True):
     if len(prefix) != k_factor:
       # we dont support words less than k factor
       return
-    preference_list = self.rangemap.nodes_for_key(prefix)
-    # TODO: Forward to other nodes if not already forwarding and replication_factor > 1
-    # right now we *only* forward if we hit the wrong node (same for remove)
-    if self.config.node_id not in preference_list and commit_to_wal:
-      print("I'm not coordinator, forwarding")
-      return self.coordinator.add(word, preference_list)
     self.sem.acquire()
-    self.wal.commit("add", word)
-
+    if commit_to_wal: self.wal.commit("add", word)
     if prefix not in self.range_tries:
       self.range_tries[prefix] = Trie()
     trie = self.range_tries[prefix]
@@ -59,20 +44,13 @@ def remove(self, word, commit_to_wal=True):
     if len(prefix) != k_factor or prefix not in self.range_tries:
       # we dont support words less than k factor
       return
-    preference_list = self.rangemap.nodes_for_key(prefix)
-    if self.config.node_id not in preference_list and commit_to_wal:
-      print("I'm not coordinator, forwarding")
-      return self.coordinator.remove(word, preference_list)
     self.sem.acquire()
-    self.wal.commit("remove", word)
+    if commit_to_wal: self.wal.commit("remove", word)
     trie = self.range_tries[prefix]
     trie.remove(word)
     self.sem.release()
     return True
-  
-  def is_coordinator_for_key(self, key):
-    return self.config.node_id in self.rangemap.nodes_for_key(key)
-  
+    
   def load(self, prefix, trie_dict):
     trie = Trie()
     trie.load(trie_dict)

todo.md

@@ -1,13 +1,9 @@
 # TODO
 
 * Distributed writes
-    * Writes should be forwarded to a shard owner
-    * If no node is healthy, return an error
-    * Get writes accepted by configured quorum before returning a result
-        * Version tries (keep a timestamp of last write on keys and share timestamp)
+    * Version tries (keep a timestamp of last write on keys and share timestamp)
         * Last write wins
-        * Send whole subtries in write forwards
-    * Expected that client will handle locking etc. and user will configure trifecta for use case
+        * Treat a subtrie as an object. Send whole trie in write forwards
 
 ---
 
@@ -29,6 +25,8 @@
 * Address todos
 * Logging + metrics + stats page
 * Limits and pagination
+* Parallel writes, aborts, hedging
+* Rollback writes if quorum fails
 
 ---------------------------
 
@@ -43,4 +41,5 @@ Ideas
 1. A shard-aware client that skips one potential hop
 1. Namespacing and/or word tagging
 1. Sorting (popularity:desc, alphabetical:desc)
-1. Concurrent healthchecks / gossips
+1. Concurrent healthchecks / gossips
+1. Expected that client will handle locking etc. and user will configure trifecta for use case (i.e. writes to triedb should be done in sequence by clients), but maybe we can handle this for them (by versioning/tombstoning words, not tries)

wal.py

@@ -28,7 +28,7 @@ def readlines(self):
     for walfile in globbed:
       with open(walfile, 'r') as f:
         for line in f.readlines():
-          action, object = line.rstrip().split(" ")
+          action, object = line.rstrip().split(" ", maxsplit=1)
           yield (action, object)
 
   def rotate(self):