exp/ingest/ledgerbackend: Wrap metaPipe in bufio.Reader (#2763)

Wraps `metaPipe` in `stellarCoreRunner` in `bufio.Reader`. This improves
the speed of reading from a meta pipe.

In 936bc3a we removed `bufio.Reader` based on an observation [1] that it
slows down reading from a pipe. However, after running a CPU profiler on
reingestion code to debug another change I realized that a lot of time
is spent on reading from a pipe in methods like `xdr.Decoder.DecodeInt`
or `xdr.Decoder.DecodeUint` (that read just 4 bytes).

When reingesting 1000 ledgers all executions of this method take 17% of
entire run time! It was clear that there's an overhead time connected to
reading from a pipe. After wrapping meta pipe in `bufio.Reader` the
speed of reingestion of the same range improved by 25% (this isn't
affected by Stellar-Core init time, the timer starts after
`PrepareRange` is done).

Before:
```
INFO[2020-06-30T12:03:44.671+02:00] Reingestion done     duration=74.352344343 from=29999000 pid=44533 service=expingest to=30000000
```
After:
```
INFO[2020-06-30T12:06:14.829+02:00] Reingestion done     duration=55.449094814 from=29999000 pid=45181 service=expingest to=30000000
```

The change in 936bc3a solves the problems with reflection used in XDR
decoder being slow. This commit mitigates issues with overhead time
added when reading from a pipe.

[1] #2552 (comment)

exp/ingest/ledgerbackend/stellar_core_runner_posix.go

@@ -3,6 +3,7 @@
 package ledgerbackend
 
 import (
+	"bufio"
 	"os"
 	"syscall"
 
@@ -43,7 +44,9 @@ func (c *stellarCoreRunner) start() error {
 		return errors.Wrap(e, "error starting stellar-core")
 	}
 
-	c.metaPipe = readFile
+	// Do not remove bufio.Reader wrapping. Turns out that each read from a pipe
+	// adds an overhead time so it's better to preload data to a buffer.
+	c.metaPipe = bufio.NewReaderSize(readFile, 1024*1024)
 	return nil
 }
 

exp/ingest/ledgerbackend/stellar_core_runner_windows.go

@@ -41,6 +41,8 @@ func (c *stellarCoreRunner) start() error {
 		return e
 	}
 
+	// Do not remove bufio.Reader wrapping. Turns out that each read from a pipe
+	// adds an overhead time so it's better to preload data to a buffer.
 	c.metaPipe = bufio.NewReaderSize(connection, 1024*1024)
 	return nil
 }

services/horizon/internal/expingest/fsm.go

@@ -577,6 +577,8 @@ func (h reingestHistoryRangeState) run(s *system) (transition, error) {
 		h.fromLedger = 2
 	}
 
+	startTime = time.Now()
+
 	if h.force {
 		if err := s.historyQ.Begin(); err != nil {
 			return stop(), errors.Wrap(err, "Error starting a transaction")
@@ -630,6 +632,12 @@ func (h reingestHistoryRangeState) run(s *system) (transition, error) {
 		}
 	}
 
+	log.WithFields(logpkg.F{
+		"from":     h.fromLedger,
+		"to":       h.toLedger,
+		"duration": time.Since(startTime).Seconds(),
+	}).Info("Reingestion done")
+
 	return stop(), nil
 }