this repo has no description
waiting for the done channel is unnecessary
+9
-13
+9
-13
events/diskpersist.go
+9
-13
events/diskpersist.go
···
46
46
}
47
47
48
48
type persistJob struct {
49
-
Buf []byte
50
-
Evt *XRPCStreamEvent
51
-
Done chan error
49
+
Buf []byte
50
+
Evt *XRPCStreamEvent
51
+
Buffer *bytes.Buffer // so we can put it back in the pool when we're done
52
52
}
53
53
54
54
type jobResult struct {
···
263
263
}
264
264
265
265
if err := p.doPersist(job); err != nil {
266
-
job.Done <- err
267
-
continue
268
-
} else {
269
-
job.Done <- nil
266
+
log.Error(err)
270
267
}
271
268
272
269
if len(p.evtbuf) > 50 {
···
325
322
326
323
for _, ej := range p.evtbuf {
327
324
p.broadcast(ej.Evt)
325
+
p.buffers.Put(ej.Buffer)
328
326
}
329
327
330
328
p.evtbuf = p.evtbuf[:0]
···
375
373
376
374
func (p *DiskPersistence) Persist(ctx context.Context, e *XRPCStreamEvent) error {
377
375
buffer := p.buffers.Get().(*bytes.Buffer)
378
-
defer p.buffers.Put(buffer)
379
376
380
377
buffer.Truncate(0)
381
378
···
404
401
binary.LittleEndian.PutUint32(b[4:], evtKind)
405
402
binary.LittleEndian.PutUint32(b[8:], uint32(len(b)-headerSize))
406
403
407
-
done := make(chan error, 1)
408
404
p.evts <- &persistJob{
409
-
Buf: b,
410
-
Evt: e,
411
-
Done: done,
405
+
Buf: b,
406
+
Evt: e,
407
+
Buffer: buffer,
412
408
}
413
409
414
-
return <-done // NB: getting rid of this channel 'wait for things to be done' thing makes it go a good deal faster
410
+
return nil
415
411
}
416
412
417
413
type evtHeader struct {