Update limits

commonspace/object/tree/objecttree/rawloader.go

@@ -275,7 +275,7 @@ func (r *rawChangeLoader) loadAppendEntry(id string) (entry rawCacheEntry, err e
 	}
 	size := len(rawChange.RawChange)
 	r.buf = rawChange.RawChange
-	
+
 	change, err := r.changeBuilder.Unmarshall(rawChange, false)
 	if err != nil {
 		return

commonspace/sync/guard.go

@@ -3,22 +3,29 @@ package sync
 import "sync"
 
 type guard struct {
-	mu    sync.Mutex
+	mu         sync.Mutex
-	taken map[string]struct{}
+	taken      map[string]struct{}
+	limit      int
+	takenCount int
 }
 
-func newGuard() *guard {
+func newGuard(limit int) *guard {
 	return &guard{
 		taken: make(map[string]struct{}),
+		limit: limit,
 	}
 }
 
 func (g *guard) TryTake(id string) bool {
 	g.mu.Lock()
 	defer g.mu.Unlock()
+	if g.limit != 0 && g.takenCount >= g.limit {
+		return false
+	}
 	if _, exists := g.taken[id]; exists {
 		return false
 	}
+	g.takenCount++
 	g.taken[id] = struct{}{}
 	return true
 }
@@ -26,5 +33,6 @@ func (g *guard) TryTake(id string) bool {
 func (g *guard) Release(id string) {
 	g.mu.Lock()
 	defer g.mu.Unlock()
+	g.takenCount--
 	delete(g.taken, id)
 }

commonspace/sync/requestmanager.go

@@ -40,7 +40,7 @@ func NewRequestManager(handler syncdeps.SyncHandler, metric syncdeps.QueueSizeUp
 	return &requestManager{
 		requestPool:   NewRequestPool(),
 		handler:       handler,
-		incomingGuard: newGuard(),
+		incomingGuard: newGuard(1000),
 		metric:        metric,
 	}
 }

commonspace/sync/requestpool.go

@@ -27,7 +27,7 @@ func NewRequestPool() RequestPool {
 		ctx:       ctx,
 		cancel:    cancel,
 		pools:     make(map[string]*tryAddQueue),
-		peerGuard: newGuard(),
+		peerGuard: newGuard(0),
 	}
 }
 
@@ -57,7 +57,7 @@ func (rp *requestPool) QueueRequestAction(peerId, objectId string, action func(c
 	)
 	pool, exists = rp.pools[peerId]
 	if !exists {
-		pool = newTryAddQueue(100, 100)
+		pool = newTryAddQueue(10, 100)
 		rp.pools[peerId] = pool
 		pool.Run()
 	}