test (3a9ca3dc) · Commits · jan.koester / blogi

plugins/media/media.cpp

+40 −1

Original line number	Diff line number	Diff line
		@@ -1323,8 +1323,15 @@ namespace blogi {
		Args->config->buildurl("media/getimage/"))==0){

		auto *ss = _getStreamState(req);
		if(!ss \|\| !ss->client)
		if(!ss \|\| !ss->client) {
		std::cerr << "[STREAM-DIAG] ts=" << _diagNowMs() << " ptr=" << &req
		<< " event=no_state" << std::endl;
		return false;
		}

		std::cerr << "[STREAM-DIAG] ts=" << _diagNowMs() << " ptr=" << &req
		<< " event=enter produced=" << ss->produced
		<< " sendSize=" << ss->sendSize << std::endl;

		if(ss->produced < ss->sendSize){
		static constexpr size_t STREAM_CHUNK = 65536;
		@@ -1337,9 +1344,18 @@ namespace blogi {
		size_t n;
		try {
		n = ss->client->readBodyChunkNonBlocking(chunk.data(), toRead);
		} catch(std::exception &e) {
		std::cerr << "[STREAM-DIAG] ts=" << _diagNowMs() << " ptr=" << &req
		<< " event=read_exception what=" << e.what() << std::endl;
		n = static_cast<size_t>(-1);
		} catch(...) {
		std::cerr << "[STREAM-DIAG] ts=" << _diagNowMs() << " ptr=" << &req
		<< " event=read_exception what=unknown" << std::endl;
		n = static_cast<size_t>(-1);
		}
		std::cerr << "[STREAM-DIAG] ts=" << _diagNowMs() << " ptr=" << &req
		<< " event=read n=" << (n == static_cast<size_t>(-1) ? -1 : (long long)n)
		<< std::endl;
		if(n == static_cast<size_t>(-1)){
		std::cerr << "media: upstream stream error after "
		<< ss->produced << "/" << ss->sendSize << " bytes" << std::endl;
		@@ -1372,8 +1388,14 @@ namespace blogi {
		}
		}

		std::cerr << "[STREAM-DIAG] ts=" << _diagNowMs() << " ptr=" << &req
		<< " event=exit produced=" << ss->produced
		<< " sendSize=" << ss->sendSize << std::endl;

		if(ss->produced >= ss->sendSize){
		ss->client.reset();
		std::cerr << "[STREAM-DIAG] ts=" << _diagNowMs() << " ptr=" << &req
		<< " event=complete" << std::endl;
		return false;
		}
		return true;
		@@ -1394,6 +1416,9 @@ namespace blogi {
		std::unique_lock lock(_streamMutex);
		auto it = _streams.find(reinterpret_cast<uintptr_t>(&req));
		if(it != _streams.end()){
		std::cerr << "[STREAM-DIAG] ts=" << _diagNowMs() << " ptr=" << &req
		<< " event=disconnect produced=" << it->second.produced
		<< " sendSize=" << it->second.sendSize << std::endl;
		it->second.client.discard();
		_streams.erase(it);
		}
		@@ -1894,10 +1919,22 @@ namespace blogi {
		// call _reapStaleStreamsLocked() again to release them back.
		static constexpr std::chrono::seconds STREAM_IDLE_TIMEOUT{30};

		static long long _diagNowMs(){
		return std::chrono::duration_cast<std::chrono::milliseconds>(
		std::chrono::system_clock::now().time_since_epoch()).count();
		}

		void _reapStaleStreamsLocked(){
		const auto now = std::chrono::steady_clock::now();
		for(auto it = _streams.begin(); it != _streams.end();){
		if(now - it->second.last_activity > STREAM_IDLE_TIMEOUT){
		std::cerr << "[STREAM-DIAG] ts=" << _diagNowMs() << " ptr=0x"
		<< std::hex << it->first << std::dec
		<< " event=reaped idle_ms="
		<< std::chrono::duration_cast<std::chrono::milliseconds>(
		now - it->second.last_activity).count()
		<< " produced=" << it->second.produced
		<< " sendSize=" << it->second.sendSize << std::endl;
		// Connection may hold un-drained bytes from the abandoned
		// response — discard rather than returning it to the pool,
		// where reuse would desync the next reader.
		@@ -1934,6 +1971,8 @@ namespace blogi {
		size_t sendSize, int tid){
		std::unique_lock lock(_streamMutex);
		_reapStaleStreamsLocked();
		std::cerr << "[STREAM-DIAG] ts=" << _diagNowMs() << " ptr=" << &req
		<< " event=stored sendSize=" << sendSize << std::endl;
		auto &ss = _streams[reinterpret_cast<uintptr_t>(&req)];
		ss.client = std::move(client);
		ss.sendSize = sendSize;