virtual directory colocation

requirements.txt

@@ -1,3 +1,3 @@
 requests
-pyarrow
+pyarrow<16.0.0
 boto3

src/lib.go

@@ -83,14 +83,19 @@ func (s byScore) Less(i, j int) bool {
 	return bytes.Compare(s[i].score, s[j].score) == 1
 }
 
-func key2volume(key []byte, volumes []string, count int, svcount int) []string {
+func key2volume(key []byte, volumes []string, count int, svcount int, vdir_colocation bool) []string {
 	// this is an intelligent way to pick the volume server for a file
 	// stable in the volume server name (not position!)
 	// and if more are added the correct portion will move (yay md5!)
 	var sortvols []sortvol
 	for _, v := range volumes {
 		hash := md5.New()
-		hash.Write(key)
+		// truncate key at last slash if vdir_colocation is enabled
+		if i := bytes.LastIndexByte(key, '/'); vdir_colocation && i != -1 {
+			hash.Write(key[:i])
+		} else {
+			hash.Write(key)
+		}
 		hash.Write([]byte(v))
 		score := hash.Sum(nil)
 		sortvols = append(sortvols, sortvol{score, v})

src/lib_test.go

@@ -31,7 +31,7 @@ func Test_key2volume(t *testing.T) {
 		"blah":       "curly",
 	}
 	for k, v := range tests {
-		ret := key2volume([]byte(k), volumes, 1, 3)
+		ret := key2volume([]byte(k), volumes, 1, 3, false)
 		if strings.Split(ret[0], "/")[0] != v {
 			t.Fatal("key2volume function broke", k, ret, v)
 		}

src/main.go

@@ -22,14 +22,15 @@ type App struct {
 	lock  map[string]struct{}
 
 	// params
-	uploadids  map[string]bool
-	volumes    []string
-	fallback   string
-	replicas   int
-	subvolumes int
-	protect    bool
-	md5sum     bool
-	voltimeout time.Duration
+	uploadids       map[string]bool
+	volumes         []string
+	fallback        string
+	replicas        int
+	subvolumes      int
+	protect         bool
+	md5sum          bool
+	voltimeout      time.Duration
+	vdir_colocation bool
 }
 
 func (a *App) UnlockKey(key []byte) {
@@ -77,6 +78,7 @@ func main() {
 	verbose := flag.Bool("v", false, "Verbose output")
 	md5sum := flag.Bool("md5sum", true, "Calculate and store MD5 checksum of values")
 	voltimeout := flag.Duration("voltimeout", 1*time.Second, "Volume servers must respond to GET/HEAD requests in this amount of time or they are considered down, as duration")
+	vdir_colocation := flag.Bool("vdircolocation", false, "Enable virtual directory colocation, which places items with the same prefix before the last slash on the same volume server(s)")
 	flag.Parse()
 
 	volumes := strings.Split(*pvolumes, ",")
@@ -110,15 +112,16 @@ func main() {
 
 	fmt.Printf("volume servers: %s\n", volumes)
 	a := App{db: db,
-		lock:       make(map[string]struct{}),
-		uploadids:  make(map[string]bool),
-		volumes:    volumes,
-		fallback:   *fallback,
-		replicas:   *replicas,
-		subvolumes: *subvolumes,
-		protect:    *protect,
-		md5sum:     *md5sum,
-		voltimeout: *voltimeout,
+		lock:            make(map[string]struct{}),
+		uploadids:       make(map[string]bool),
+		volumes:         volumes,
+		fallback:        *fallback,
+		replicas:        *replicas,
+		subvolumes:      *subvolumes,
+		protect:         *protect,
+		md5sum:          *md5sum,
+		voltimeout:      *voltimeout,
+		vdir_colocation: *vdir_colocation,
 	}
 
 	if command == "server" {

src/rebalance.go

@@ -135,7 +135,7 @@ func (a *App) Rebalance() {
 		key := make([]byte, len(iter.Key()))
 		copy(key, iter.Key())
 		rec := toRecord(iter.Value())
-		kvolumes := key2volume(key, a.volumes, a.replicas, a.subvolumes)
+		kvolumes := key2volume(key, a.volumes, a.replicas, a.subvolumes, a.vdir_colocation)
 		wg.Add(1)
 		reqs <- RebalanceRequest{
 			key:      key,

src/rebuild.go

@@ -41,7 +41,7 @@ func rebuild(a *App, vol string, name string) bool {
 		return false
 	}
 
-	kvolumes := key2volume(key, a.volumes, a.replicas, a.subvolumes)
+	kvolumes := key2volume(key, a.volumes, a.replicas, a.subvolumes, a.vdir_colocation)
 
 	if !a.LockKey(key) {
 		fmt.Println("lockKey issue", key)

src/server.go

@@ -143,7 +143,7 @@ func (a *App) Delete(key []byte, unlink bool) int {
 
 func (a *App) WriteToReplicas(key []byte, value io.Reader, valuelen int64) int {
 	// we don't have the key, compute the remote URL
-	kvolumes := key2volume(key, a.volumes, a.replicas, a.subvolumes)
+	kvolumes := key2volume(key, a.volumes, a.replicas, a.subvolumes, a.vdir_colocation)
 
 	// push to leveldb initially as deleted, and without a hash since we don't have it yet
 	if !a.PutRecord(key, Record{kvolumes, SOFT, ""}) {
@@ -221,7 +221,7 @@ func (a *App) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 			// fall through to fallback
 			remote = fmt.Sprintf("http://%s%s", a.fallback, key)
 		} else {
-			kvolumes := key2volume(key, a.volumes, a.replicas, a.subvolumes)
+			kvolumes := key2volume(key, a.volumes, a.replicas, a.subvolumes, a.vdir_colocation)
 			if needs_rebalance(rec.rvolumes, kvolumes) {
 				w.Header().Set("Key-Balance", "unbalanced")
 				fmt.Println("on wrong volumes, needs rebalance")
@@ -369,7 +369,7 @@ func (a *App) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 			return
 		}
 
-		kvolumes := key2volume(key, a.volumes, a.replicas, a.subvolumes)
+		kvolumes := key2volume(key, a.volumes, a.replicas, a.subvolumes, a.vdir_colocation)
 		rbreq := RebalanceRequest{key: key, volumes: rec.rvolumes, kvolumes: kvolumes}
 		if !rebalance(a, rbreq) {
 			w.WriteHeader(400)