chore: experiment with a simpler protocol #715
Conversation
Summary of ChangesHello @olavloite, I'm Gemini Code Assist1! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed! This pull request introduces an experimental, simpler socket-based protocol designed to facilitate Spanner operations. It includes a new Go server that exposes Spanner functionality over a socket connection, along with client-side libraries in both Go and .NET to interact with this server. The aim is to investigate the viability and benefits of this alternative communication layer for future development, potentially offering improved performance or integration flexibility. Highlights
🧠 New Feature in Public Preview: You can now enable Memory to help Gemini Code Assist learn from your team's feedback. This makes future code reviews more consistent and personalized to your project's style. Click here to enable Memory in your admin console. Using Gemini Code AssistThe full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips. Invoking Gemini You can request assistance from Gemini at any point by creating a comment using either
Customization To customize Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a Limitations & Feedback Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counter productive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for Github and other Google products, sign up here. You can also get AI-powered code generation, chat, as well as code reviews directly in the IDE at no cost with the Gemini Code Assist IDE Extension. Footnotes
|
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces an experimental socket-based communication protocol as an alternative to gRPC. The changes are extensive, adding a Go-based socket server, a Go client library, and a C# client implementation, along with the necessary build scripts. While this is a significant and promising piece of work, my review has identified several critical issues that need to be addressed. These include race conditions in the Go server, nil pointer dereferences during protocol decoding, and incorrect or incomplete implementations in the C# client, particularly around asynchronous operations and thread safety. I've also provided some medium-severity suggestions to improve code maintainability and fix minor bugs. Addressing these points will be crucial for the stability and correctness of this new protocol.
| func createBeginMessage(reader *bufio.Reader) (*BeginMessage, error) { | ||
| msg := &BeginMessage{} | ||
| options, err := protocol.ReadTransactionOptions(reader) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| msg.Options = options | ||
| return msg, nil | ||
| } | ||
|
|
||
| func createCommitMessage(reader *bufio.Reader) (*CommitMessage, error) { | ||
| return &CommitMessage{}, nil | ||
| } | ||
|
|
||
| func createRollbackMessage(reader *bufio.Reader) (*RollbackMessage, error) { | ||
| return &RollbackMessage{}, nil | ||
| } | ||
|
|
||
| func createCommitResultMessage(reader *bufio.Reader) (*CommitResultMessage, error) { | ||
| msg := &CommitResultMessage{} | ||
| resp, err := protocol.ReadCommitResponse(reader) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| msg.Response = resp | ||
|
|
||
| return msg, nil | ||
| } |
There was a problem hiding this comment.
Several create... functions (createBeginMessage, createCommitMessage, createRollbackMessage, createCommitResultMessage) do not initialize the messageId field of the embedded message struct. This will cause the MessageId() method to return a zero value, which is incorrect and will likely cause issues when the message is processed. Each of these functions should initialize the messageId with the corresponding message ID constant.
For example, createCommitMessage should be:
func createCommitMessage(reader *bufio.Reader) (*CommitMessage, error) {
return &CommitMessage{message: message{messageId: CommitMessageId}}, nil
}| var status *spb.Status | ||
| if err := proto.Unmarshal(b, status); err != nil { | ||
| return nil, err | ||
| } | ||
| res.Status = status |
There was a problem hiding this comment.
proto.Unmarshal is called with a nil pointer for the status variable. This will cause a runtime error. You need to initialize status with a pointer to a spb.Status struct.
| var status *spb.Status | |
| if err := proto.Unmarshal(b, status); err != nil { | |
| return nil, err | |
| } | |
| res.Status = status | |
| var status spb.Status | |
| if err := proto.Unmarshal(b, &status); err != nil { | |
| return nil, err | |
| } | |
| res.Status = &status |
| connectionHandler.handler = &messageHandler{ | ||
| conn: connectionHandler, | ||
| } | ||
| s.handlers = append(s.handlers, connectionHandler) |
There was a problem hiding this comment.
Appending to s.handlers slice from multiple goroutines without a mutex is a race condition. This can lead to data corruption or panics under load. You should add a sync.Mutex to the Server struct and use it to protect access to s.handlers.
Also, the handlers slice is never read. If it's not intended to be used, it should be removed to avoid confusion and the overhead of locking.
| private static Task ExecuteStartupAsync(NetworkStream stream, PoolImpl pool, CancellationToken cancellationToken) | ||
| { | ||
| var startup = new StartupMessage(pool); | ||
| return startup.WriteAsync(stream, cancellationToken); | ||
| } |
There was a problem hiding this comment.
The ExecuteStartupAsync method is incomplete. It writes the startup message but does not flush the stream or wait for a status response from the server. This will cause the connection to fail to establish correctly. The async method should mirror the logic of the synchronous ExecuteStartup method, which includes flushing the stream and reading the status message. This will require creating an async version of ReadStatusMessage.
| private readonly Dictionary<Pool, PoolImpl> _pools = new(); | ||
| private readonly Dictionary<Connection, ConnectionImpl> _connections = new(); |
There was a problem hiding this comment.
The _pools and _connections dictionaries are not thread-safe. ISpannerLib implementations are expected to be thread-safe, as they can be used concurrently. Using System.Collections.Generic.Dictionary without synchronization can lead to race conditions. You should use System.Collections.Concurrent.ConcurrentDictionary instead.
private readonly System.Collections.Concurrent.ConcurrentDictionary<Pool, PoolImpl> _pools = new();
private readonly System.Collections.Concurrent.ConcurrentDictionary<Connection, ConnectionImpl> _connections = new();| return err | ||
| } | ||
| if m.status != nil { | ||
| b, _ := proto.Marshal(m.status) |
There was a problem hiding this comment.
The error from proto.Marshal is being ignored. While it's unlikely to fail for a status.Status protobuf, it's best practice to handle all errors.
| b, _ := proto.Marshal(m.status) | |
| b, err := proto.Marshal(m.status) | |
| if err != nil { | |
| return err | |
| } |
| func WriteValue(writer *bufio.Writer, value *structpb.Value) error { | ||
| switch value.Kind.(type) { | ||
| case *structpb.Value_NullValue: | ||
| if err := WriteByte(writer, 0); err != nil { | ||
| return err | ||
| } | ||
| case *structpb.Value_BoolValue: | ||
| if err := WriteByte(writer, 1); err != nil { | ||
| return err | ||
| } | ||
| if err := WriteBool(writer, value.GetBoolValue()); err != nil { | ||
| return err | ||
| } | ||
| case *structpb.Value_NumberValue: | ||
| if err := WriteByte(writer, 2); err != nil { | ||
| return err | ||
| } | ||
| if err := WriteFloat64(writer, value.GetNumberValue()); err != nil { | ||
| return err | ||
| } | ||
| case *structpb.Value_StringValue: | ||
| if err := WriteByte(writer, 3); err != nil { | ||
| return err | ||
| } | ||
| if err := WriteString(writer, value.GetStringValue()); err != nil { | ||
| return err | ||
| } | ||
| case *structpb.Value_ListValue: | ||
| if err := WriteByte(writer, 4); err != nil { | ||
| return err | ||
| } | ||
| // Write the number of values and then each value. | ||
| if err := WriteInt32(writer, int32(len(value.GetListValue().GetValues()))); err != nil { | ||
| return err | ||
| } | ||
| for _, value := range value.GetListValue().GetValues() { | ||
| if err := WriteValue(writer, value); err != nil { | ||
| return err | ||
| } | ||
| } | ||
| case *structpb.Value_StructValue: | ||
| return status.Errorf(codes.Unimplemented, "struct value not yet supported") | ||
| } | ||
| return nil | ||
| } |
There was a problem hiding this comment.
The WriteValue function uses magic numbers (0, 1, 2, 3, 4) for ValueType. These should be replaced with the ValueType constants defined in this file (e.g., ValueTypeNull, ValueTypeBool) to improve readability and maintainability.
func WriteValue(writer *bufio.Writer, value *structpb.Value) error {
switch value.Kind.(type) {
case *structpb.Value_NullValue:
if err := WriteByte(writer, byte(ValueTypeNull)); err != nil {
return err
}
case *structpb.Value_BoolValue:
if err := WriteByte(writer, byte(ValueTypeBool)); err != nil {
return err
}
if err := WriteBool(writer, value.GetBoolValue()); err != nil {
return err
}
case *structpb.Value_NumberValue:
if err := WriteByte(writer, byte(ValueTypeNumber)); err != nil {
return err
}
if err := WriteFloat64(writer, value.GetNumberValue()); err != nil {
return err
}
case *structpb.Value_StringValue:
if err := WriteByte(writer, byte(ValueTypeString)); err != nil {
return err
}
if err := WriteString(writer, value.GetStringValue()); err != nil {
return err
}
case *structpb.Value_ListValue:
if err := WriteByte(writer, byte(ValueTypeList)); err != nil {
return err
}
// Write the number of values and then each value.
if err := WriteInt32(writer, int32(len(value.GetListValue().GetValues()))); err != nil {
return err
}
for _, value := range value.GetListValue().GetValues() {
if err := WriteValue(writer, value); err != nil {
return err
}
}
case *structpb.Value_StructValue:
return status.Errorf(codes.Unimplemented, "struct value not yet supported")
}
return nil
}| return nil, err | ||
| } | ||
| if numParams > 0 { | ||
| params = &structpb.Struct{} |
There was a problem hiding this comment.
The params variable is re-assigned here, but it was already initialized on line 11. This is redundant. The same issue exists in ReadParamTypes.
|
|
||
| internal class RowsImpl : Rows | ||
| { | ||
| internal RowsImpl Create(ConnectionImpl connection) |
There was a problem hiding this comment.
The Create method is defined as an instance method, but it functions as a factory. It should be a static method.
internal static RowsImpl Create(ConnectionImpl connection)| public override ListValue? Next() | ||
| { | ||
| var hasMoreRows = Encoding.ReadBool(_stream); | ||
| if (!hasMoreRows) | ||
| { | ||
|
|
||
| } | ||
| } |
There was a problem hiding this comment.
The Next() method is incomplete. It reads hasMoreRows but then has an empty if block and does not return a value. This method needs to be fully implemented to read and return row data or handle the end of the result set.
1 participant
No description provided.