diranged
commented
6 years ago I've found where the bug lies.. and have a partial fix. The bug is that if handleReplicationMsg() returns an error, it immediately halts the replicationLoop() goroutine. When that is halted, and because the replicationMessages channel has size: 0, the main connectReplicateLoop() routine halts when trying to insert the next message into the channel.
The failure process looks like this ...
connectReplicateLoop()starts up and pushesreplicationMessages <- msg1- No blocking occurs because
replicationLoop()was already up and waiting for a message replicationLoop()processesmsg1and sees its too big (> 1mb message), so immediatelyreturnsconnectReplicateLoop()has already run back around in a circle though and is trying to pushreplicationMessages <- msg2- Deadlock
diff --git a/pg_kinesis.go b/pg_kinesis.go
index ebbfdf1..f982340 100644
--- a/pg_kinesis.go
+++ b/pg_kinesis.go
@@ -458,6 +458,8 @@ func replicationLoop(replicationMessages chan *pgx.ReplicationMessage, replicati
if err != nil {
replicationFinished <- err // already wrapped
+ logf("error received, dumped into replicationFinished: %s", err)
+ logf("replicationFinished size: %d", len(replicationFinished))
return
}
}
@@ -619,6 +621,7 @@ func fetchLag(slot *string, conn *pgx.Conn) error {
func pkeyLagLoop(slot *string, nonReplConn *pgx.Conn, replicationFinished chan error) {
for !done.IsSet() {
+ logf("channel length: %d", len(replicationFinished))
err := fetchPKs(nonReplConn)
if err != nil {
replicationFinished <- err
@@ -694,28 +697,34 @@ func connectReplicateLoop(slot *string, sourceConfig pgx.ConnConfig, stream *str
return errors.Wrap(err, "waiting for replication message failed")
}
- // check if the replicating goroutine died
- select {
- case replErr = <-replicationFinished:
- default:
- }
-
- if replErr != nil {
- logf("stopping replication due to replication goroutine failure")
- return replErr // already wrapped
- }
if message != nil {
if message.WalMessage != nil {
// this is not exactly the server time
// but we are taking over this field as PG does not send it down
message.WalMessage.ServerTime = uint64(now.UnixNano())
- replicationMessages <- message
+
+ // check if the replicating goroutine died
+ select {
+ case replicationMessages <- message:
+ case replErr = <-replicationFinished:
+ }
+
+ logf("done")
} else if message.ServerHeartbeat != nil {
keepaliveRequested = message.ServerHeartbeat.ReplyRequested == 1
}
}
+
+ if replErr != nil {
+ logf("stopping replication due to replication goroutine failure")
+ return replErr // already wrapped
+ }
+
err = sendKeepalive(conn, keepaliveRequested)
if err != nil {
return errors.Wrap(err, "unable to send keepalive")The problem is that even with this patch, things aren't great. As soon as the connectReplicateLoop() dies off, the main() loop sleeps, then restarts (and there is a debate on whether or not that should happen). The problem is that when it restarts, it starts the conenctReplicateLoop() creates all new output channels, including new ones to send more data to Kinesis. At this point, you end up with two different pkeyLagLoop() routines running!
At this point, everything is hosed.. there are multiple functions running trying to write to common global variables (pks.Store... and the stats struct). You also, for some reason that I have not yet figured out, can only run through the connectReplicateLoop() once before it deadlocks when trying to add a message to its new replicationMessages channel.
It seems like at this point, the app just needs to die?
taybin
First off, this tool is great .. and I really want to see it succeed... so with that said, we're testing it out and finding that after it runs for a while, it deadlocks in our environment. This is the second time we've seen this in 24 hrs.
At this point, its been locked like this for ~10 hours .. so its clearly dead. It keeps outputting the status line, but nothing else. Any suggestions on how to troubleshoot?