offline-deal connection rest by peer

[wuqinqiang]

Checklist

• This is not a question or a support request. If you have any boost related questions, please ask in the discussion forum.
• This is not a new feature or enhancement request. If it is, please open a new idea discussion instead. New feature and enhancement requests would be entertained by the boost team after a thorough discussion only.
• I have searched on the issue tracker and the discussion forum, and there is no existing related issue or discussion.
• I am running the Latest release, or the most recent RC(release canadiate) for the upcoming release or the dev branch(master), or have an issue updating to any of these.
• I did not make any code changes to boost.

Boost component

• boost daemon - storage providers
• boost client
• boost UI
• boost data-transfer
• boost index-provider
• Other

Boost Version

boost --version

boost version 1.6.0-rc1+git.739d7bb

Describe the Bug

I wanted to send deals concurrently, so i reused the "dealDCmdAction" function in the deal_cmd file(without modifying any code inside it).

The pseudocode is as follows:

        var wg sync.WaitGroup
	for i := 0; i < 10; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			dealCmdAction()
		}()
	}
	wg.Wait()

After run program, only the first few orders send to the same sp were successful, and the rest failed.

Then, i added the following code to "dealCmdAction",

boost/cmd/boost/deal_cmd.go

Line 133 in a4a9bfb

	defer n.Host.Close()

This solved the problem. What happened?

Logging Information

As above

Repo Steps

As above

[LexLuthr]

Would it be possible to provide access to your full code?
At first glance, it looks like you are not closing any streams that you open. So, either connManager or other security/resource manager are shutting down connection from other side.

Since you are throwing all the connections in goroutines, you would need to ensure to close them after response it received. Thus, you fix is the correct way to go here.

[wuqinqiang]

I apologize, providing the complete code is not very convenient, but the code above fully reflects my actual operation.

I completely copied the dealCmdAction function, except that I added the defer n.Host.Close() when I solved the problem.

I mean, why didn't you add defer n.Host.Close() to dealCmdAction?

The only difference I can see is that the official program executes the command once, boost deal or offline-deal, and calls dealCmdAction once, then the program ends. Whereas I called dealCmdAction multiple times concurrently, and then the program ended.

[LexLuthr]

From the code snippet, it doesn't look like you are using a context in dealCmdAction(). If so then how are you passing the context to n.Host.NewStream()?

[wuqinqiang]

more code

func forkcctx(cctx *cli.Context, deal *deal) (ctx *cli.Context, err error) {
	//new a flag
	f := newFlag(cctx)
	
	// new a context 
	ctx = cli.NewContext(cli.NewApp(), f, nil)

	if err := ctx.Set("filename", deal.FileName); err != nil {
		return nil, fmt.Errorf("filename err:%v", err)
	}
	if err := ctx.Set("payload-cid", deal.PayloadCid); err != nil {
		return nil, fmt.Errorf("payload-cid err:%v", err)
	}
	if err := ctx.Set("piece-size", deal.PieceSize); err != nil {
		return nil, fmt.Errorf("piece-size err:%v", err)
	}
	if err := ctx.Set("car-size", deal.CarSize); err != nil {
		return nil, fmt.Errorf("car-size err:%v", err)
	}
	if err := ctx.Set("commp", deal.PieceCid); err != nil {
		return nil, fmt.Errorf("commp err:%v", err)
	}

	if err := ctx.Set("provider", deal.Provider); err != nil {
		return nil, fmt.Errorf("provider err:%v", err)
	}

	if err := ctx.Set("start-epoch",deal.staetEpoch); err != nil {
		return nil, fmt.Errorf("start-epoch err:%v", err)
	}

	if err := ctx.Set("duration", deal.duration); err != nil {
		return nil, fmt.Errorf("duration err:%v", err)
	}
	return
}

func newFlag(cctx *cli.Context) *flag.FlagSet {
	flagSet := flag.NewFlagSet("deal", flag.ContinueOnError)
	
	flagSet.Bool("verified", false, "verified")
	flagSet.String("filename", "", "filename")
	flagSet.String("payload-cid", "", "payload-cid")
	flagSet.String("piece-size", "", "piece-size")
	flagSet.String("car-size", "", "car-size")
	flagSet.String("commp", "", "commp")
	flagSet.String("provider", "", "provider")
	flagSet.String("storage-price", "0", "storage-price")
	flagSet.Int("start-epoch", 0, "start-epoch")
	flagSet.Int("duration", 518400, "duration")

	flagSet.String("repo", cctx.String("repo"), "repo")
	flagSet.String("wallet", cctx.String("wallet"), "wallet")
	flagSet.String("json", cctx.String("json"), "json")
	return flagSet
}

func main() {
	var wg sync.WaitGroup
	for i := 0; i < 10; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			forkctx:=forkcctx(cctx)
			dealCmdAction(forkctx)
		}()
	}
	wg.Wait()

}

each call dealCmdAction is a new ctx.

[LexLuthr]

https://github.com/filecoin-project/boost/blame/490e23ceb76c0d5d26cad0d69c69c7a8755dce09/cmd/boost/deal_cmd.go#L285 closes the new stream in the code as well. Are you sure code you did not make any other changes in the code.

It would not be possible to help without the code itself.

[beck-8]

My scenario is this: I wrote a python parallel sealing script and encountered the following error.

failed to negotiate stream multiplexer:read tcp xx

Under the same concurrency conditions, I encountered less of this in v1.5.1, quite a lot in 1.6.0-rc1, less in 1.6.0-rc2, but much more than in 1.5.1.
The environments I tested were all based on a basically new test network to keep the factors under control.

[LexLuthr]

Where is this script running? Does the script on-board data or make deals? Please provide details and possibly link to the code? There is no way to diagnose issues without the access to custom code.

[beck-8]

This script only sends offline transactions, and I put it on the boostd machine, so no network issues arise. I have no custom code, I just call the binary in Python.

def boost_deal(payloadcid, piecesize, carsize, commp, filename):
    status, output = subprocess.getstatusoutput(
        f'./boost --json offline-deal --verified=true --wallet={from_add}  --payload-cid={payloadcid} --piece-size={piecesize} --car-size={carsize} --commp={commp} --provider={provider} --duration={duration} --storage-price=0 --start-epoch={start_epoch}')
    return status, output

[beck-8]

My concurrency is 10-40, which means maybe 10-40 transactions are sent at the same time.

[dirkmc]

We should create an JSON RPC API similar to BoostDummyDeal so that users can create deals directly against the JSON RPC API instead of having to connect through libp2p.

[LexLuthr]

That seems like a good approach to bypass all the libp2p things. Let me create an issue and get started.