CODEWITHSUNDEEP
gogrpc
Abhijet
Abhijet

Reputation: 165

grpc server stops receiving messages after sending many messages simultaneously

I am implementing a simple grpc service where the summary of a task is to be sent to the grpc server. Everything works fine if I send less number of messages but when I begin to send like 5000 messages the server stops and gets deadline exceeded message in client side. I also tried to reconnect again but found the error message as.

rpc error: code = Unavailable desc = all SubConns are in TransientFailure, latest connection error: timed out waiting for server handshake

The server shows no error and is alive.

I tried setting GRPC_GO_REQUIRE_HANDSHAKE=off as well but the error still prevails. I also implemented sending summary in batch but same scenerio repeated.

Is there any limitations to number of messages to be sent in grpc?

Here is my service proto


// The Result service definition.
service Result {
  rpc ConntectMaster(ConnectionRequest) returns (stream ExecutionCommand) {}
  rpc postSummary(Summary) returns(ExecutionCommand) {}
}

message Summary{

  int32 successCount = 1;
  int32 failedCount = 2;
  int32 startTime = 3;
  repeated TaskResult results = 4;
  bool isLast = 5;
  string id = 6;
}

postSummary implementation in sever

// PostSummary posts the summary to the master
func (server *Server) PostSummary(ctx context.Context, in *pb.Summary) (*pb.ExecutionCommand, error) {

    for i := 0; i < len(in.Results); i++ {

        res := in.Results[i]
        log.Printf("%s --> %d Res :: %s, len : %d", in.Id, i, res.Id, len(in.Results))

    }
    return &pb.ExecutionCommand{Type: stopExec}, nil
}

func postSummaryInBatch(executor *Executor, index int) {
    summary := pb.Summary{
        SuccessCount: int32(executor.summary.successCount),
        FailedCount:  int32(executor.summary.failedCount),
        Results:      []*pb.TaskResult{},
        IsLast:       false,
    }

    if index >= len(executor.summary.TaskResults) {
        summary.IsLast = true
        return
    }

    var to int
    batch := 500
    if (index + batch) <= len(executor.summary.TaskResults) {
        to = index + batch
    } else {
        to = len(executor.summary.TaskResults)
    }
    for i := index; i < to; i++ {
        result := executor.summary.TaskResults[i]
        taskResult := pb.TaskResult{
            Id:   result.id,
            Msg:  result.msg,
            Time: result.time,
        }
        // log.Printf("adding res : %s ", taskResult.Id)

        if result.err != nil {
            taskResult.IsError = true
        }
        summary.Results = append(summary.Results, &taskResult)
    }
    summary.Id = fmt.Sprintf("%d-%d", index, to)
    log.Printf("sent from  %d to %d ", index, to)
    postSummary(executor, &summary, 0)
    postSummaryInBatch(executor, to)
}

func postSummary(executor *Executor, summary *pb.Summary, retryCount int) {
    ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
    defer cancel()

    cmd, err := client.PostSummary(ctx, summary)
    if err != nil {
        if retryCount < 3 {
            reconnect(executor)
            postSummary(executor, summary, retryCount+1)
        }
        log.Printf(err.Error())
        // log.Fatal("cannot send summary report")
    } else {
        processServerCommand(executor, cmd)
    }
}

Upvotes: 0

Views: 1506

Answers (1)

owlwalks
owlwalks

Reputation: 1681

grpc default maxReceiveMessageSize is 4MB, your grpc client probably went over that limit.

grpc uses h2 in transport layer which opens only one tcp conn and multiplex "requests" over that, reduce significant overhead compare to h1, I wouldn't worry too much for batching and will just make individual calls to grpc server.

Upvotes: 1

Related Questions