filer: avoid possible timeouts for updates and deletions

1 files changed, 45 insertions, 0 deletions

diff --git a/weed/util/queue_unbounded.go b/weed/util/queue_unbounded.go
new file mode 100644
index 000000000..664cd965e
--- /dev/null
+++ b/weed/util/queue_unbounded.go
@@ -0,0 +1,45 @@
+package util
+
+import "sync"
+
+type UnboundedQueue struct {
+	outbound     []string
+	outboundLock sync.RWMutex
+	inbound      []string
+	inboundLock  sync.RWMutex
+}
+
+func NewUnboundedQueue() *UnboundedQueue {
+	q := &UnboundedQueue{}
+	return q
+}
+
+func (q *UnboundedQueue) EnQueue(items ...string) {
+	q.inboundLock.Lock()
+	defer q.inboundLock.Unlock()
+
+	q.outbound = append(q.outbound, items...)
+
+}
+
+func (q *UnboundedQueue) Consume(fn func([]string)) {
+	q.outboundLock.Lock()
+	defer q.outboundLock.Unlock()
+
+	if len(q.outbound) == 0 {
+		q.inboundLock.Lock()
+		inbountLen := len(q.inbound)
+		if inbountLen > 0 {
+			t := q.outbound
+			q.outbound = q.inbound
+			q.inbound = t
+		}
+		q.inboundLock.Unlock()
+	}
+
+	if len(q.outbound) > 0 {
+		fn(q.outbound)
+		q.outbound = q.outbound[:0]
+	}
+
+}