(maxBatchSize int)
| 487 | } |
| 488 | |
| 489 | func (peer *Peer) RoutineSequentialSender(maxBatchSize int) { |
| 490 | device := peer.device |
| 491 | defer func() { |
| 492 | defer device.Log.Verbosef("%v - Routine: sequential sender - stopped", peer) |
| 493 | peer.stopping.Done() |
| 494 | }() |
| 495 | device.Log.Verbosef("%v - Routine: sequential sender - started", peer) |
| 496 | |
| 497 | bufs := make([][]byte, 0, maxBatchSize) |
| 498 | parBufs := make([][]byte, 0, maxBatchSize) |
| 499 | eps := make([]conn.Endpoint, 0) |
| 500 | parEps := make([]conn.Endpoint, 0) |
| 501 | |
| 502 | for elemsContainer := range peer.queue.outbound.c { |
| 503 | bufs = bufs[:0] |
| 504 | eps = eps[:0] |
| 505 | parBufs = parBufs[:0] |
| 506 | parEps = parEps[:0] |
| 507 | if elemsContainer == nil { |
| 508 | return |
| 509 | } |
| 510 | if !peer.isRunning.Load() { |
| 511 | // peer has been stopped; return re-usable Elems to the shared pool. |
| 512 | // This is an optimization only. It is possible for the peer to be stopped |
| 513 | // immediately after this check, in which case, elem will get processed. |
| 514 | // The timers and SendBuffers code are resilient to a few stragglers. |
| 515 | // TODO: rework peer shutdown order to ensure |
| 516 | // that we never accidentally keep timers alive longer than necessary. |
| 517 | elemsContainer.Lock() |
| 518 | for _, elem := range elemsContainer.elems { |
| 519 | device.PutMessageBuffer(elem.buffer) |
| 520 | device.PutOutboundElement(elem) |
| 521 | } |
| 522 | device.PutOutboundElementsContainer(elemsContainer) |
| 523 | continue |
| 524 | } |
| 525 | dataSent := false |
| 526 | elemsContainer.Lock() |
| 527 | for _, elem := range elemsContainer.elems { |
| 528 | if len(elem.packet) != MessageKeepaliveSize { |
| 529 | dataSent = true |
| 530 | } |
| 531 | bufs = append(bufs, elem.packet) |
| 532 | eps = append(eps, elem.endpoint) |
| 533 | } |
| 534 | |
| 535 | peer.timersAnyAuthenticatedPacketTraversal(false) |
| 536 | peer.timersAnyAuthenticatedPacketSent() |
| 537 | |
| 538 | err := partitionBuffersByEndpoint(bufs, eps, &parBufs, &parEps, peer) |
| 539 | if err != nil { |
| 540 | device.Log.Errorf("%v - Failed to send data packets: %v", peer, err) |
| 541 | continue |
| 542 | } |
| 543 | err = peer.SendBuffers(parBufs, parEps) |
| 544 | if dataSent { |
| 545 | peer.timersDataSent() |
| 546 | } |
no test coverage detected