Skip to content

Commit 4f70798

Browse files
fbivillefbiville
committed
Backport new 5.0 implementation of connection timeout hints
1 parent 7af52e3 commit 4f70798

5 files changed

Lines changed: 319 additions & 43 deletions

File tree

‎neo4j/error.go‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,8 @@
2020
package neo4j
2121

2222
import (
23+
"context"
24+
"errors"
2325
"fmt"
2426
"io"
2527
"net"
@@ -125,7 +127,9 @@ func wrapError(err error) error {
125127
if err == nil {
126128
return nil
127129
}
128-
if err == io.EOF {
130+
if err == io.EOF ||
131+
errors.Is(err, context.DeadlineExceeded) ||
132+
errors.Is(err, context.Canceled) {
129133
return &ConnectivityError{inner: err}
130134
}
131135
switch e := err.(type) {

‎neo4j/internal/bolt/dechunker.go‎

Lines changed: 49 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -20,36 +20,48 @@
2020
package bolt
2121

2222
import (
23+
"context"
2324
"encoding/binary"
25+
rio "github.com/neo4j/neo4j-go-driver/v4/neo4j/internal/racingio"
2426
"github.com/neo4j/neo4j-go-driver/v4/neo4j/log"
25-
"io"
2627
"net"
2728
"time"
2829
)
2930

3031
// dechunkMessage takes a buffer to be reused and returns the reusable buffer
3132
// (might have been reallocated to handle growth), the message buffer and
3233
// error.
33-
// If a non-default connection read timeout configuration hint is passed, the dechunker resets the connection read
34-
// deadline as well after successfully reading a chunk (NOOP messages included)
35-
func dechunkMessage(conn net.Conn, msgBuf []byte, readTimeout time.Duration,
36-
logger log.Logger, logName, logId string) ([]byte, []byte, error) {
34+
// Reads will race against the provided context ctx
35+
// If the server provides the connection read timeout hint readTimeout, a new context will be created from that timeout
36+
// and the user-provided context ctx before every read
37+
func dechunkMessage(
38+
conn net.Conn,
39+
msgBuf []byte,
40+
readTimeout time.Duration,
41+
logger log.Logger,
42+
logName string,
43+
logId string) ([]byte, []byte, error) {
44+
3745
sizeBuf := []byte{0x00, 0x00}
3846
off := 0
3947

48+
reader := rio.NewRacingReader(conn)
49+
4050
for {
41-
_, err := io.ReadFull(conn, sizeBuf)
51+
updatedCtx, cancelFunc := newContext(readTimeout, logger, logName, logId)
52+
_, err := reader.ReadFull(updatedCtx, sizeBuf)
4253
if err != nil {
4354
return msgBuf, nil, err
4455
}
56+
if cancelFunc != nil { // reading has been completed, time to release the context
57+
cancelFunc()
58+
}
4559
chunkSize := int(binary.BigEndian.Uint16(sizeBuf))
4660
if chunkSize == 0 {
4761
if off > 0 {
4862
return msgBuf, msgBuf[:off], nil
4963
}
5064
// Got a nop chunk
51-
resetConnectionReadDeadline(conn, readTimeout, logger,
52-
logName, logId)
5365
continue
5466
}
5567

@@ -60,20 +72,42 @@ func dechunkMessage(conn net.Conn, msgBuf []byte, readTimeout time.Duration,
6072
msgBuf = newMsgBuf
6173
}
6274
// Read the chunk into buffer
63-
_, err = io.ReadFull(conn, msgBuf[off:(off+chunkSize)])
75+
updatedCtx, cancelFunc = newContext(readTimeout, logger, logName, logId)
76+
_, err = reader.ReadFull(updatedCtx, msgBuf[off:(off+chunkSize)])
6477
if err != nil {
6578
return msgBuf, nil, err
6679
}
80+
if cancelFunc != nil { // reading has been completed, time to release the context
81+
cancelFunc()
82+
}
6783
off += chunkSize
68-
resetConnectionReadDeadline(conn, readTimeout, logger, logName, logId)
6984
}
7085
}
7186

72-
func resetConnectionReadDeadline(conn net.Conn, readTimeout time.Duration, logger log.Logger, logName, logId string) {
73-
if readTimeout < 0 {
74-
return
87+
// newContext computes a new context and cancel function if a readTimeout is set
88+
func newContext(
89+
readTimeout time.Duration,
90+
logger log.Logger,
91+
logName string,
92+
logId string) (context.Context, context.CancelFunc) {
93+
94+
ctx := context.Background()
95+
if readTimeout >= 0 {
96+
newCtx, cancelFunc := context.WithTimeout(ctx, readTimeout)
97+
logger.Debugf(logName, logId,
98+
"read timeout of %s applied, chunk read deadline is now: %s",
99+
readTimeout.String(),
100+
deadlineOf(newCtx),
101+
)
102+
return newCtx, cancelFunc
75103
}
76-
if err := conn.SetReadDeadline(time.Now().Add(readTimeout)); err != nil {
77-
logger.Error(logName, logId, err)
104+
return ctx, nil
105+
}
106+
107+
func deadlineOf(ctx context.Context) string {
108+
deadline, hasDeadline := ctx.Deadline()
109+
if !hasDeadline {
110+
return "N/A (no deadline set)"
78111
}
112+
return deadline.String()
79113
}

‎neo4j/internal/bolt/dechunker_test.go‎

Lines changed: 13 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,7 @@ package bolt
2222
import (
2323
"bytes"
2424
"encoding/binary"
25+
"github.com/neo4j/neo4j-go-driver/v4/neo4j/log"
2526
"net"
2627
"reflect"
2728
"testing"
@@ -108,17 +109,10 @@ func TestDechunker(t *testing.T) {
108109

109110
func TestDechunkerWithTimeout(ot *testing.T) {
110111
timeout := time.Millisecond * 600
111-
serv, cli := net.Pipe()
112-
defer func() {
113-
AssertNoError(ot, serv.Close())
114-
AssertNoError(ot, cli.Close())
115-
}()
116-
AssertNoError(ot, serv.SetReadDeadline(time.Now().Add(timeout)))
117-
logger := &noopLogger{}
118-
logName := "dechunker"
119-
logId := "dechunker-test"
120112

121113
ot.Run("Resets connection deadline upon successful reads", func(t *testing.T) {
114+
serv, cli := net.Pipe()
115+
defer closePipe(ot, serv, cli)
122116
go func() {
123117
time.Sleep(timeout / 2)
124118
AssertWriteSucceeds(t, cli, []byte{0x00, 0x00})
@@ -128,32 +122,24 @@ func TestDechunkerWithTimeout(ot *testing.T) {
128122
AssertWriteSucceeds(t, cli, []byte{0x00, 0x00})
129123
}()
130124
buffer := make([]byte, 2)
131-
_, _, err := dechunkMessage(serv, buffer, timeout, logger, logName,
132-
logId)
125+
_, _, err := dechunkMessage(serv, buffer, timeout, log.Void{}, "", "")
133126
AssertNoError(t, err)
134127
AssertTrue(t, reflect.DeepEqual(buffer, []byte{0xCA, 0xFE}))
135128
})
136129

137130
ot.Run("Fails when connection deadline is reached", func(t *testing.T) {
138-
_, _, err := dechunkMessage(serv, nil, timeout, logger, logName,
139-
logId)
140-
AssertError(t, err)
141-
AssertStringContain(t, err.Error(), "read pipe")
142-
})
143-
144-
}
145-
146-
type noopLogger struct {
147-
}
131+
serv, cli := net.Pipe()
132+
defer closePipe(ot, serv, cli)
148133

149-
func (*noopLogger) Error(string, string, error) {
150-
}
134+
_, _, err := dechunkMessage(serv, nil, timeout, log.Void{}, "", "")
151135

152-
func (*noopLogger) Warnf(string, string, string, ...interface{}) {
153-
}
136+
AssertError(t, err)
137+
AssertStringContain(t, err.Error(), "context deadline exceeded")
138+
})
154139

155-
func (*noopLogger) Infof(string, string, string, ...interface{}) {
156140
}
157141

158-
func (*noopLogger) Debugf(string, string, string, ...interface{}) {
142+
func closePipe(t *testing.T, srv, cli net.Conn) {
143+
AssertNoError(t, srv.Close())
144+
AssertNoError(t, cli.Close())
159145
}

‎neo4j/internal/racingio/reader.go‎

Lines changed: 99 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,99 @@
1+
/*
2+
* Copyright (c) "Neo4j"
3+
* Neo4j Sweden AB [http://neo4j.com]
4+
*
5+
* This file is part of Neo4j.
6+
*
7+
* Licensed under the Apache License, Version 2.0 (the "License");
8+
* you may not use this file except in compliance with the License.
9+
* You may obtain a copy of the License at
10+
*
11+
* http://www.apache.org/licenses/LICENSE-2.0
12+
*
13+
* Unless required by applicable law or agreed to in writing, software
14+
* distributed under the License is distributed on an "AS IS" BASIS,
15+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16+
* See the License for the specific language governing permissions and
17+
* limitations under the License.
18+
*/
19+
20+
package racingio
21+
22+
import (
23+
"context"
24+
"errors"
25+
"fmt"
26+
"io"
27+
)
28+
29+
type RacingReader interface {
30+
Read(ctx context.Context, bytes []byte) (int, error)
31+
ReadFull(ctx context.Context, bytes []byte) (int, error)
32+
}
33+
34+
func NewRacingReader(reader io.Reader) RacingReader {
35+
return &racingReader{reader: reader}
36+
}
37+
38+
type racingReader struct {
39+
reader io.Reader
40+
}
41+
42+
func (rr *racingReader) Read(ctx context.Context, bytes []byte) (int, error) {
43+
return rr.race(ctx, bytes, read)
44+
}
45+
46+
func (rr *racingReader) ReadFull(ctx context.Context, bytes []byte) (int, error) {
47+
return rr.race(ctx, bytes, readFull)
48+
}
49+
50+
func (rr *racingReader) race(ctx context.Context, bytes []byte, readFn func(io.Reader, []byte) (int, error)) (int, error) {
51+
if err := ctx.Err(); err != nil {
52+
return 0, wrapRaceError(err)
53+
}
54+
resultChan := make(chan *ioResult, 1)
55+
defer close(resultChan)
56+
go func() {
57+
n, err := readFn(rr.reader, bytes)
58+
defer func() {
59+
// When the read operation completes, the outer function may have returned already.
60+
// In that situation, the channel will have been closed and the result emission will crash.
61+
// Let's just swallow the panic that may happen and ignore it
62+
_ = recover()
63+
}()
64+
resultChan <- &ioResult{
65+
n: n,
66+
err: err,
67+
}
68+
}()
69+
select {
70+
case <-ctx.Done():
71+
return 0, wrapRaceError(ctx.Err())
72+
case result := <-resultChan:
73+
return result.n, wrapRaceError(result.err)
74+
}
75+
}
76+
77+
func wrapRaceError(err error) error {
78+
if err == nil {
79+
return nil
80+
}
81+
if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
82+
// temporary adjustment for 4.x
83+
return fmt.Errorf("i/o timeout: %w", err)
84+
}
85+
return err
86+
}
87+
88+
type ioResult struct {
89+
n int
90+
err error
91+
}
92+
93+
func read(reader io.Reader, bytes []byte) (int, error) {
94+
return reader.Read(bytes)
95+
}
96+
97+
func readFull(reader io.Reader, bytes []byte) (int, error) {
98+
return io.ReadFull(reader, bytes)
99+
}

0 commit comments

Comments
 (0)