RejectedSoftware Forums

On Sun, 04 Sep 2016 16:45:43 GMT, Alexey Kulentsov wrote:

Hi all!
How to deal with such type of error? No my code in trace, no clue what happened and where to see.

Task terminated with unhandled exception: Acquiring reader of already owned connection.

Ok, I resolve it, but it was not good time.
http://vibed.org/api/vibe.http.server/HTTPServerResponse.waitForConnectionClose
I tried to use this function to close my service threads on HTTP connection close. It was a bad idea because when this function running, you can't write to response bodyWriter. I found in documentation nothing about this exclusive usage.

On Tue, 06 Sep 2016 06:42:23 GMT, Jack Applegame wrote:

On Tue, 06 Sep 2016 00:28:57 GMT, Alexey Kulentsov wrote:

On Sun, 04 Sep 2016 16:45:43 GMT, Alexey Kulentsov wrote:
Ok, I resolve it, but it was not good time.
http://vibed.org/api/vibe.http.server/HTTPServerResponse.waitForConnectionClose
I tried to use this function to close my service threads on HTTP connection close. It was a bad idea because when this function running, you can't write to response bodyWriter. I found in documentation nothing about this exclusive usage.
I'm pretty sure that sharing the connection between the tasks is impossible.

It's not about sharing connection, because when I replace waitForConnectionClose() for explicit waiting loop then problem disappears. Now this method looks like:

void setExitByHTTPClose(HTTPServerResponse response)
{
	runTask((){
			while(response.connected())
				sleep(5.seconds);
			if(stillRunning)
				writer.send(WriterCommand.JustExit);
	});
}

On Tue, 06 Sep 2016 12:29:13 GMT, Jack Applegame wrote:

You said

It was a bad idea because when this function running, you can't write to response bodyWriter.
But how you can write to bodyWriter when running waitForConnectionClose() without creating parallel task and sharing connection between them?
However, I believe that waitForConnectionClose() should not take ownership.

Could you post your previous problematic code, please.

This is part of web interface:

void getChanneldown(HTTPServerRequest request, HTTPServerResponse response)
{
	import ChannelD;
	
	auto controller = getController(request);
	
	scope channel = new ChannelD;
	channel.setExitByHTTPClose(response);
	
	controller.subscribe(channel.writer);
	scope(exit) controller.unsubscribe();
	
	channel.run(response.bodyWriter);

}

This is method ChannelD.run:

/**
 * Run download channel
 */
void run(OutputStream response)
{
	// running flag
	running = true;
	scope(exit) running = false;
	
	// output with flush
	void echo(string message)
	{
		response.write(message);
		response.flush;
	}
	
	// send all messages to client
	void sendAll()
	{
		debug
		{
			import std.stdio: writeln;
			writeln("send to stream:", messagesToSend.serializeToJson.toString);
		}
		
		echo(messagesToSend.serializeToJson.toString);
	}
	
	// if mesages exists, just send it end exit
	if(messagesToSend.length)
	{
		sendAll();
		return;
	}
	
	// waiting loop will be in this task
	writer = Task.getThis();
	
	// forced drop connection after interval
	if(channel_reload_interval.total!"seconds" > 0)
		runTask((){
				sleep(channel_reload_interval);
				if(stillRunning)
					writer.send(WriterCommand.FlushAndExit);
		});
	
	// run pinger to keep connection alive
	auto pinger = runTask((){
		while(stillRunning){
			writer.send(WriterCommand.Ping);
			sleep(ping_interval);
		}
	});
	
	// run main writer
	bool exit = false;	
	
	while(!exit) receive(
			(WriterCommand cmd){
				with(WriterCommand) final switch(cmd)
				{
					case Ping: 
						echo(" ");
						break;
						
					case FlushAndExit: 
						sendAll();
						goto case;
						
					case JustExit:
						exit = true; 
						break;
				}
			},
			(ChannelMessage message){
				messagesToSend ~= message;
				sendAll();
				exit = true;
			}
			
	);
	response.finalize();
}

}

And now setExitByHTTPClose():

void setExitByHTTPClose(HTTPServerResponse response)
{
	runTask((){
			while(response.connected())
				sleep(5.seconds);
			if(stillRunning)
				writer.send(WriterCommand.JustExit);
	});
}

There delegate in another Task uses response object. Now there is not error.
before if was like

runTask((){
		while(!response.waitForConnectionClose()){}
		if(stillRunning)
			writer.send(WriterCommand.JustExit);
});

and it was error described in my first message.

On Wed, 07 Sep 2016 06:50:04 GMT, Jack Applegame wrote:

I don't see anything wrong. You start three tasks: pinger, closed connection watcher and channel worker. But only one task (channel worker) sends data to the stream.
Looks like a bug.

I made repo with minimal code to experiment:

git clone https://github.com/crimaniak/waitForConn.git

I hope Sönke will explain what is wrong here.