diff options
| -rwxr-xr-x | worker.go | 27 | ||||
| -rwxr-xr-x | worker_watcher.go | 3 |
2 files changed, 16 insertions, 14 deletions
@@ -61,7 +61,7 @@ type WorkerEvent struct { var pool = sync.Pool{ New: func() interface{} { buf := make([]byte, 10240) - return buf + return &buf }, } @@ -320,13 +320,18 @@ func (w *WorkerProcess) Kill() error { return nil } -func (w *WorkerProcess) put(data []byte) { - data = make([]byte, 10240) +// put the pointer, to not allocate new slice +// but erase it len and then return back +func (w *WorkerProcess) put(data *[]byte) { + *data = (*data)[:0] + *data = (*data)[:cap(*data)] + pool.Put(data) } -func (w *WorkerProcess) get() []byte { - return pool.Get().([]byte) +// get pointer to the byte slice +func (w *WorkerProcess) get() *[]byte { + return pool.Get().(*[]byte) } // Write appends the contents of pool to the errBuffer, growing the errBuffer as @@ -338,22 +343,22 @@ func (w *WorkerProcess) watch() { case <-w.stop: buf := w.get() // read the last data - n, _ := w.rd.Read(buf[:]) - w.events.Push(WorkerEvent{Event: EventWorkerLog, Worker: w, Payload: buf[:n]}) + n, _ := w.rd.Read(*buf) + w.events.Push(WorkerEvent{Event: EventWorkerLog, Worker: w, Payload: (*buf)[:n]}) w.mu.Lock() // write new message - w.stderr.Write(buf[:n]) + w.stderr.Write((*buf)[:n]) w.mu.Unlock() w.put(buf) return default: // read the max 10kb of stderr per one read buf := w.get() - n, _ := w.rd.Read(buf[:]) - w.events.Push(WorkerEvent{Event: EventWorkerLog, Worker: w, Payload: buf[:n]}) + n, _ := w.rd.Read(*buf) + w.events.Push(WorkerEvent{Event: EventWorkerLog, Worker: w, Payload: (*buf)[:n]}) w.mu.Lock() // write new message - w.stderr.Write(buf[:n]) + w.stderr.Write((*buf)[:n]) w.mu.Unlock() w.put(buf) } diff --git a/worker_watcher.go b/worker_watcher.go index f9b3d372..f8fb67a9 100755 --- a/worker_watcher.go +++ b/worker_watcher.go @@ -228,7 +228,6 @@ func (ww *workerWatcher) GetFreeWorker(ctx context.Context) (WorkerBase, error) if w == nil { continue } - //ww.ReduceWorkersCount() return w, nil case <-ctx.Done(): return nil, errors.E(op, errors.NoFreeWorkers, errors.Str("no free workers in the stack, timeout exceed")) @@ -236,7 +235,6 @@ func (ww *workerWatcher) GetFreeWorker(ctx context.Context) (WorkerBase, error) } } - //ww.ReduceWorkersCount() return w, nil } @@ -278,7 +276,6 @@ func (ww *workerWatcher) RemoveWorker(wb WorkerBase) error { wb.State().Set(StateRemove) return nil - } // O(1) operation |