transfer: clean up contract creation to create fewer contracts in parallel

Author: xcolwell

transfer.go

@@ -96,7 +96,7 @@ func DefaultClientSettingsNoNetworkEvents() *ClientSettings {
 func DefaultSendBufferSettings() *SendBufferSettings {
 	return &SendBufferSettings{
 		CreateContractTimeout:       30 * time.Second,
-		CreateContractRetryInterval: 5 * time.Second,
+		CreateContractRetryInterval: 15 * time.Second,
 		MinResendInterval:           1 * time.Second,
 		MaxResendInterval:           5 * time.Second,
 		// no backoff
@@ -1442,6 +1442,14 @@ func (self *SendSequence) Run() {
 				sendContract.ackedByteCount,
 				sendContract.unackedByteCount,
 			)
+			// flush queued contracts for already sent contracts
+			// contractKey = ContractKey{
+			// 	Destination:       sendContract.path.DestinationMask(),
+			// 	IntermediaryIds:   self.intermediaryIds,
+			// 	CompanionContract: self.companionContract,
+			// 	ForceStream:       self.forceStream,
+			// }
+			// self.client.ContractManager().FlushContractQueue(contractKey, true)
 		}
 
 		// drain the buffer
@@ -1682,8 +1690,7 @@ func (self *SendSequence) Run() {
 func (self *SendSequence) updateContract(messageByteCount ByteCount) bool {
 	// `sendNoContract` is a mutual configuration
 	// both sides must configure themselves to require no contract from each other
-	isStream := self.destination.IsStream() || 0 < self.intermediaryIds.Len()
-	if !isStream && self.client.ContractManager().SendNoContract(self.destination.DestinationId) {
+	if self.client.ContractManager().SendNoContract(self.destination.DestinationId) {
 		return true
 	}
 	if self.sendContract != nil && self.sendContract.update(messageByteCount) {
@@ -1774,11 +1781,15 @@ func (self *SendSequence) updateContract(messageByteCount ByteCount) bool {
 			}
 		}
 
-		if traceNextContract(0) {
-			return true
+		endTime := time.Now().Add(self.sendBufferSettings.CreateContractTimeout)
+
+		if self.sendContract != nil {
+			// there should be a queued up contract
+			if traceNextContract(min(self.sendBufferSettings.CreateContractTimeout, self.sendBufferSettings.CreateContractRetryInterval)) {
+				return true
+			}
 		}
 
-		endTime := time.Now().Add(self.sendBufferSettings.CreateContractTimeout)
 		for {
 			select {
 			case <-self.ctx.Done():
@@ -3193,7 +3204,7 @@ func (self *ReceiveSequence) updateContract(item *receiveItem) bool {
 	}
 	// `receiveNoContract` is a mutual configuration
 	// both sides must configure themselves to require no contract from each other
-	if !self.source.IsStream() && self.client.ContractManager().ReceiveNoContract(self.source.SourceId) {
+	if self.client.ContractManager().ReceiveNoContract(self.source.SourceId) {
 		return true
 	}
 	return false

transfer_contract_manager.go

@@ -321,15 +321,15 @@ func (self *ContractManager) contractStatus(contractStatus *ContractStatus) {
 func (self *ContractManager) Receive(source TransferPath, frames []*protocol.Frame, provideMode protocol.ProvideMode) {
 	if source.IsControlSource() {
 		for _, frame := range frames {
-			self.handleControlFrame(frame)
+			self.handleControlFrame(nil, frame)
 		}
 	}
 }
 
-func (self *ContractManager) handleControlFrame(frame *protocol.Frame) error {
+func (self *ContractManager) handleControlFrame(createContractKey *ContractKey, frame *protocol.Frame) error {
 	switch frame.MessageType {
 	case protocol.MessageType_TransferCreateContractResult:
-		contracts, contractErrors := self.parseControlFrame(frame)
+		contracts, contractErrors := self.parseControlFrame(createContractKey, frame)
 		for contractKey, contract := range contracts {
 			c := func() error {
 				err := self.addContract(contractKey, contract)
@@ -400,68 +400,71 @@ func (self *ContractManager) handleControlFrame(frame *protocol.Frame) error {
 }
 
 // frames are verified before calling to be from source ControlId
-func (self *ContractManager) parseControlFrame(frame *protocol.Frame) (
+func (self *ContractManager) parseControlFrame(createContractKey *ContractKey, frame *protocol.Frame) (
 	contracts map[ContractKey]*protocol.Contract,
 	contractErrors map[ContractKey]protocol.ContractError,
 ) {
 	contracts = map[ContractKey]*protocol.Contract{}
 	contractErrors = map[ContractKey]protocol.ContractError{}
 
 	addResult := func(v *protocol.CreateContractResult) {
-		contractKey := ContractKey{}
-		if v.CreateContract != nil {
-			contractKey.CompanionContract = v.CreateContract.Companion
-			if v.CreateContract.ForceStream != nil {
-				contractKey.ForceStream = *v.CreateContract.ForceStream
+		var contractKey *ContractKey
+		if createContractKey != nil {
+			contractKey = createContractKey
+		} else if createContract := v.CreateContract; createContract != nil {
+			contractKey = &ContractKey{}
+			var err error
+			contractKey.Destination, err = TransferPathFromBytes(
+				nil,
+				createContract.DestinationId,
+				nil,
+			)
+			if err != nil {
+				return
 			}
-			if v.CreateContract.IntermediaryIds != nil {
-				if intermediaryIds, err := MultiHopIdFromBytes(v.CreateContract.IntermediaryIds); err == nil {
+			contractKey.CompanionContract = createContract.Companion
+			if createContract.ForceStream != nil {
+				contractKey.ForceStream = *createContract.ForceStream
+			}
+			if createContract.IntermediaryIds != nil {
+				if intermediaryIds, err := MultiHopIdFromBytes(createContract.IntermediaryIds); err == nil {
 					contractKey.IntermediaryIds = intermediaryIds
 				}
 			}
 		}
 
 		if contractError := v.Error; contractError != nil {
-			if v.CreateContract != nil {
-				var err error
-				contractKey.Destination, err = TransferPathFromBytes(
-					nil,
-					v.CreateContract.DestinationId,
-					v.CreateContract.StreamId,
-				)
-				if err != nil {
-					return
-				}
+			if contractKey != nil {
+				contractErrors[*contractKey] = *contractError
+			} else {
+				glog.Infof("[contract]error with unassociated contract = %s\n", contractError)
 			}
-			contractErrors[contractKey] = *contractError
 		} else if contract := v.Contract; contract != nil {
 			storedContract := &protocol.StoredContract{}
 			err := ProtoUnmarshal(contract.StoredContractBytes, storedContract)
 			if err != nil {
 				return
 			}
 
-			if v.CreateContract != nil {
-				var err error
-				contractKey.Destination, err = TransferPathFromBytes(
-					nil,
-					v.CreateContract.DestinationId,
-					v.CreateContract.StreamId,
-				)
-				if err != nil {
-					return
-				}
-			} else {
+			if contractKey == nil && self.settings.LegacyCreateContract {
+				// this only makes sense for legacy contracts
+				contractKey = &ContractKey{}
 				contractKey.Destination, err = TransferPathFromBytes(
 					nil,
 					storedContract.DestinationId,
-					storedContract.StreamId,
+					nil,
 				)
 				if err != nil {
 					return
 				}
 			}
-			contracts[contractKey] = contract
+
+			if contractKey != nil {
+				contracts[*contractKey] = contract
+			} else {
+				// this contract can't be associated (TODO close it)
+				glog.Errorf("[contract]unassociated contract %s\n", Id(storedContract.ContractId))
+			}
 		}
 	}
 
@@ -806,18 +809,11 @@ func (self *ContractManager) addContract(contractKey ContractKey, contract *prot
 		return err
 	}
 
-	path, err := TransferPathFromBytes(
-		storedContract.SourceId,
-		storedContract.DestinationId,
-		storedContract.StreamId,
-	)
-	if err != nil {
-		return err
+	if Id(storedContract.SourceId) != self.client.ClientId() {
+		return fmt.Errorf("Contract source must be this client: %s<>%s", Id(storedContract.SourceId), self.client.ClientId())
 	}
 
-	if !path.IsStream() && path.SourceId != self.client.ClientId() {
-		return fmt.Errorf("Contract source must be this client: %s<>%s", path.SourceId, self.client.ClientId())
-	}
+	glog.V(1).Infof("[contract]add %s %s\n", self.client.ClientId(), contractKey.Destination)
 
 	func() {
 		contractQueue := self.openContractQueue(contractKey)
@@ -838,7 +834,6 @@ func (self *ContractManager) CreateContract(contractKey ContractKey, contractSeq
 	createContract := &protocol.CreateContract{
 		DestinationId:     contractKey.Destination.DestinationId.Bytes(),
 		IntermediaryIds:   contractKey.IntermediaryIds.Bytes(),
-		StreamId:          contractKey.Destination.StreamId.Bytes(),
 		TransferByteCount: uint64(self.contractByteCount(contractSeqIndex, minByteCount)),
 		Companion:         contractKey.CompanionContract,
 		ForceStream:       &contractKey.ForceStream,
@@ -852,11 +847,16 @@ func (self *ContractManager) CreateContract(contractKey ContractKey, contractSeq
 		glog.Infof("[contract]could not create contract frame = %s", err)
 		return
 	}
+
+	glog.V(1).Infof("[contract]create %s %s\n", self.client.ClientId(), contractKey.Destination)
+
 	self.client.ClientOob().SendControl(
 		[]*protocol.Frame{frame},
 		func(resultFrames []*protocol.Frame, err error) {
 			if err == nil {
-				self.Receive(SourceId(ControlId), resultFrames, protocol.ProvideMode_Network)
+				for _, resultFrame := range resultFrames {
+					self.handleControlFrame(&contractKey, resultFrame)
+				}
 			} else {
 				select {
 				case <-self.client.Done():
@@ -978,11 +978,16 @@ func (self *ContractManager) Flush(resetUsedContractIds bool) []Id {
 		self.mutex.Lock()
 		defer self.mutex.Unlock()
 
+		glog.V(1).Infof("[contract]flush %s %s\n", self.client.ClientId(), maps.Keys(self.destinationContracts))
+
 		contracts := []*protocol.Contract{}
-		for _, contractQueue := range self.destinationContracts {
+		for contractKey, contractQueue := range self.destinationContracts {
 			for _, contract := range contractQueue.Flush(resetUsedContractIds) {
 				contracts = append(contracts, contract)
 			}
+			if contractQueue.IsDone() {
+				delete(self.destinationContracts, contractKey)
+			}
 		}
 		return contracts
 	}()
@@ -1119,7 +1124,7 @@ func (self *contractQueue) Add(contract *protocol.Contract, storedContract *prot
 		glog.V(2).Infof("[contract]add update existing %s\n", contractId)
 		self.contracts[contractId] = contract
 		self.updateMonitor.NotifyAll()
-	} else if !self.usedContractIds[contractId] {
+	} else if !self.trackUsedContracts || !self.usedContractIds[contractId] {
 		glog.V(2).Infof("[contract]add %s\n", contractId)
 		if self.trackUsedContracts {
 			self.usedContractIds[contractId] = true

transfer_test.go

@@ -73,6 +73,7 @@ func TestSendReceiveSenderReset(t *testing.T) {
 	// clientSettingsA.ReceiveBufferSettings.AckBufferSize = 0
 	clientSettingsA.ForwardBufferSettings.SequenceBufferSize = 0
 	clientSettingsA.ForwardBufferSettings.IdleTimeout = 180 * time.Second
+	clientSettingsA.ContractManagerSettings.LegacyCreateContract = true
 	a := NewClient(ctx, aClientId, NewNoContractClientOob(), clientSettingsA)
 	aRouteManager := a.RouteManager()
 	aContractManager := a.ContractManager()
@@ -98,6 +99,7 @@ func TestSendReceiveSenderReset(t *testing.T) {
 	// clientSettingsB.ReceiveBufferSettings.AckBufferSize = 0
 	clientSettingsB.ForwardBufferSettings.SequenceBufferSize = 0
 	clientSettingsB.ForwardBufferSettings.IdleTimeout = 180 * time.Second
+	clientSettingsB.ContractManagerSettings.LegacyCreateContract = true
 	b := NewClient(ctx, bClientId, NewNoContractClientOob(), clientSettingsB)
 	bRouteManager := b.RouteManager()
 	bContractManager := b.ContractManager()