First commit for big refactor into a Project2 daemon, much still to do, but much core stuff in place

1 files changed, 355 insertions, 0 deletions

diff --git a/p2pvr/lib/tuner.cpp b/p2pvr/lib/tuner.cpp
new file mode 100644
index 0000000..e508152
--- /dev/null
+++ b/p2pvr/lib/tuner.cpp
@@ -0,0 +1,355 @@
+#include "tuner.h"
+#include <fcntl.h>
+#include <Ice/Ice.h>
+#include <sys/ioctl.h>
+#include <poll.h>
+#include <logger.h>
+#include <misc.h>
+#include <plugable.h>
+#include <linux/dvb/frontend.h>
+#include <linux/dvb/dmx.h>
+#include <boost/crc.hpp>
+#include <boost/tuple/tuple.hpp>
+#include "fileHandle.h"
+#include "siParsers/table.h"
+
+class FrontendNotSupported : public NotSupported {
+	public:
+		FrontendNotSupported(fe_type t) : NotSupported(stringbf("Frontend not supported: %s", t))
+				{ }
+};
+
+Tuner::Tuner(const boost::filesystem::path & df) :
+	deviceFrontend(df),
+	deviceRoot(df.branch_path()),
+	timeout(20000),
+	backgroundThread(NULL)
+{
+	int fd = open(deviceFrontend.string().c_str(), O_RDWR);
+	if (fd < 0) {
+		throw P2PVR::DeviceError(deviceFrontend.string(), strerror(errno), errno);
+	}
+	try {
+		struct dvb_frontend_info fe_info;
+		if (ioctl(fd, FE_GET_INFO, &fe_info) < 0) {
+			throw P2PVR::DeviceError(deviceFrontend.string(), strerror(errno), errno);
+		}
+		frontend = FrontendPtr(FrontendLoader::createNew<FrontendNotSupported>(fe_info.type, this, fd, fe_info));
+	}
+	catch (...) {
+		close(fd);
+		throw;
+	}
+	Logger()->messagebf(LOG_INFO, "%s: Attached to %s (%s, type %s)", __PRETTY_FUNCTION__,
+			deviceRoot, frontend->Info().name, frontend->Type());
+}
+
+Tuner::~Tuner()
+{
+	while (!backgroundClients.empty()) {
+		close(backgroundClients.begin()->first);
+		backgroundClients.erase(backgroundClients.begin());
+	}
+	if (backgroundThread) {
+		backgroundThread->join();
+		delete backgroundThread;
+	}
+}
+
+void
+Tuner::TuneTo(const DVBSI::DeliveryPtr & mp, const Ice::Current&)
+{
+	frontend->TuneTo(mp);
+}
+
+int
+Tuner::GetStatus(const Ice::Current &)
+{
+	return frontend->GetStatus();
+}
+
+std::string
+Tuner::Device() const
+{
+	return deviceRoot.string();
+}
+
+int
+Tuner::OpenDemux() const
+{
+	int demux = open((deviceRoot / "demux0").string().c_str(), O_RDWR | O_NONBLOCK);
+	if (demux < 0) {
+		throw P2PVR::DeviceError(deviceRoot.string(), strerror(errno), errno);
+	}
+	return demux;
+}
+
+void
+Tuner::ScanAndSendNetworkInformation(const P2PVR::RawDataClientPrx & client, const Ice::Current & ice)
+{
+	frontend->FrequencyScan([this, &client, &ice](long) {
+			try {
+				SendNetworkInformation(client, ice);
+				return true;
+			}
+			catch (...) {
+				return false;
+			}
+		});
+}
+
+void
+Tuner::SendNetworkInformation(const P2PVR::RawDataClientPrx & client, const Ice::Current & ice)
+{
+	SendPID(0x10, client, ice);
+}
+
+void
+Tuner::SendBouquetAssociations(const P2PVR::RawDataClientPrx & client, const Ice::Current & ice)
+{
+	SendPID(0x11, client, ice);
+}
+
+void
+Tuner::SendServiceDescriptions(const P2PVR::RawDataClientPrx & client, const Ice::Current & ice)
+{
+	SendPID(0x11, client, ice);
+}
+
+void
+Tuner::SendEventInformation(const P2PVR::RawDataClientPrx & client, const Ice::Current & ice)
+{
+	SendPID(0x12, client, ice);
+}
+
+void
+Tuner::SendPID(int pid, const P2PVR::RawDataClientPrx & client, const Ice::Current & ice) const
+{
+	Logger()->messagebf(LOG_DEBUG, "%s: pid = 0x%x", __PRETTY_FUNCTION__, pid);
+
+	ice.con->createProxy(client->ice_getIdentity());
+	FileHandle demux(OpenDemux());
+	struct dmx_sct_filter_params sctFilterParams;
+	memset(&sctFilterParams, 0, sizeof(dmx_sct_filter_params));
+	sctFilterParams.pid = pid;
+	sctFilterParams.flags = DMX_IMMEDIATE_START;
+
+	if (ioctl(demux, DMX_SET_FILTER, &sctFilterParams) < 0) {
+		throw P2PVR::DeviceError("demux", strerror(errno), errno);
+	}
+
+	ReadDemuxAndSend(demux, client);
+}
+
+void
+Tuner::ReadDemuxAndSend(int demux, const P2PVR::RawDataClientPrx & client) const
+{
+	Logger()->messagebf(LOG_DEBUG, "%s: begin", __PRETTY_FUNCTION__);
+	std::vector<Ice::AsyncResultPtr> asyncs;
+	struct pollfd ufd;
+	bool exitFlag = false;
+	do {
+		// Wait for data to appear
+		memset(&ufd, 0, sizeof(pollfd));
+		ufd.fd = demux;
+		ufd.events = POLLIN;
+		if (poll(&ufd, 1, timeout) < 1) {
+			Logger()->messagebf(LOG_DEBUG, "%s: Timed out waiting for data", __PRETTY_FUNCTION__);
+			throw P2PVR::DeviceError("demux", "Timed out. Tuned to a multiplex?", 0);
+		}
+
+		// Read it
+		P2PVR::Data buf(1 << 12);
+		int nr = read(demux, &buf.front(), buf.size());
+		if (nr < 0) {
+			throw P2PVR::DeviceError("demux", strerror(errno), errno);
+		}
+		size_t n = nr;
+		buf.resize(n);
+
+		// Verify it
+		if (n < sizeof(SiTableHeader)) {
+			Logger()->messagebf(LOG_WARNING, "Received data too small to be an SI table.");
+		}
+		auto * tab = (const SiTableHeader *)(&buf.front());
+		size_t l = sizeof(SiTableHeaderBase) + HILO(tab->section_length);
+		if (n < l) {
+			Logger()->messagebf(LOG_WARNING, "Received data shorter than its defined length.");
+			continue;
+		}
+		if (n > l) {
+			Logger()->messagebf(LOG_WARNING, "Received data longer than its defined length.");
+			continue;
+		}
+		if (!crc32(buf)) {
+			Logger()->messagebf(LOG_WARNING, "Received data is corrupted (crc32 failed).");
+			continue;
+		}
+
+		asyncs.push_back(client->begin_NewData(buf));
+
+		asyncs.erase(std::remove_if(asyncs.begin(), asyncs.end(), [&exitFlag, &client](const Ice::AsyncResultPtr & a) {
+				if (a->isCompleted()) {
+					exitFlag = client->end_NewData(a);
+					return true;
+				}
+				return false;
+			}), asyncs.end());
+	} while (!exitFlag);
+	BOOST_FOREACH(const auto & a, asyncs) {
+		client->end_NewData(a);
+	}
+	Logger()->messagebf(LOG_DEBUG, "%s: end", __PRETTY_FUNCTION__);
+}
+
+int
+Tuner::StartSendingTS(const P2PVR::PacketIds & pids, const P2PVR::RawDataClientPrx & client, const Ice::Current &)
+{
+	Logger()->message(LOG_DEBUG, __PRETTY_FUNCTION__);
+	if (pids.empty()) {
+		throw P2PVR::DeviceError("demux", "Packet Id list cannot be empty", 0);
+	}
+
+	std::lock_guard<std::mutex> g(lock);
+	int demux = backgroundClients.insert(BackgroundClients::value_type(OpenDemux(), client)).first->first;
+	
+	struct dmx_pes_filter_params pesFilterParams;
+	memset(&pesFilterParams, 0, sizeof(struct dmx_pes_filter_params));
+	Logger()->messagebf(LOG_ERR, "%s: DMX_SET_PES_FILTER for pid %d", __PRETTY_FUNCTION__, pids[0]);
+	pesFilterParams.pid = pids[0];
+	pesFilterParams.input = DMX_IN_FRONTEND;
+	pesFilterParams.output = DMX_OUT_TSDEMUX_TAP;
+	pesFilterParams.pes_type = DMX_PES_OTHER;
+	pesFilterParams.flags = 0;
+
+	if (ioctl(demux, DMX_SET_PES_FILTER, &pesFilterParams) < 0) {
+		backgroundClients.erase(demux);
+		Logger()->messagebf(LOG_ERR, "%s: DMX_SET_PES_FILTER failed (%d: %s)", __PRETTY_FUNCTION__, errno, strerror(errno));
+		throw P2PVR::DeviceError("demux", strerror(errno), errno);
+	}
+
+	for (unsigned int x = 1; x < pids.size(); x += 1) {
+		__u16 p = pids[x];
+		Logger()->messagebf(LOG_ERR, "%s: DMX_ADD_PID for pid %d", __PRETTY_FUNCTION__, p);
+		if (ioctl(demux, DMX_ADD_PID, &p) < 0) {
+			backgroundClients.erase(demux);
+			Logger()->messagebf(LOG_ERR, "%s: DMX_ADD_PID failed (%d: %s)", __PRETTY_FUNCTION__, errno, strerror(errno));
+			throw P2PVR::DeviceError("demux", strerror(errno), errno);
+		}
+	}
+
+	if (ioctl(demux, DMX_START) < 0) {
+		backgroundClients.erase(demux);
+		Logger()->messagebf(LOG_ERR, "%s: DMX_START failed (%d: %s)", __PRETTY_FUNCTION__, errno, strerror(errno));
+		throw P2PVR::DeviceError("demux", strerror(errno), errno);
+	}
+
+	startSenderThread();
+	return demux;
+}
+
+void
+Tuner::StopSendingTS(int handle, const Ice::Current &)
+{
+	Logger()->message(LOG_DEBUG, __PRETTY_FUNCTION__);
+	std::lock_guard<std::mutex> g(lock);
+	if (backgroundClients.find(handle) != backgroundClients.end()) {
+		close(handle);
+		backgroundClients.erase(handle);
+	}
+}
+
+void
+Tuner::startSenderThread()
+{
+	if (!backgroundThread) {
+		backgroundThread = new std::thread(&Tuner::senderThread, this);
+	}
+}
+
+void
+Tuner::senderThread()
+{
+	lock.lock();
+	typedef boost::tuple<P2PVR::RawDataClientPrx, Ice::AsyncResultPtr, int> AsyncCall; 
+	std::vector<AsyncCall> asyncs;
+	while (!backgroundClients.empty()) {
+		int n = backgroundClients.rbegin()->first + 1;
+		fd_set rfds;
+		FD_ZERO(&rfds);
+		BOOST_FOREACH(const auto & c, backgroundClients) {
+			FD_SET(c.first, &rfds);
+		}
+		lock.unlock();
+
+		struct timeval tv { 2, 0 };
+		switch (select(n, &rfds, NULL, NULL, &tv)) {
+			case -1: // error
+			case 0: // nothing to read, but all is well
+				break;
+			default:
+				{ // stuff to do
+					std::lock_guard<std::mutex> g(lock);
+					BOOST_FOREACH(const auto & c, backgroundClients) {
+						if (FD_ISSET(c.first, &rfds)) {
+							// Read it
+							P2PVR::Data buf(1 << 16);
+							int nr = read(c.first, &buf.front(), buf.size());
+							if (nr < 0) {
+								close(c.first);
+								backgroundClients.erase(c.first);
+								break; // backgroundClients has changed, bailout and start again
+							}
+							size_t n = nr;
+							buf.resize(n);
+							// Send it
+							asyncs.push_back(AsyncCall(c.second, c.second->begin_NewData(buf), c.first));
+							//c.second->NewData(buf);
+						}
+					}
+				}
+				break;
+		}
+		// Clean up finished async requests
+		asyncs.erase(std::remove_if(asyncs.begin(), asyncs.end(), [this](const AsyncCall & a) {
+				try {
+					if (a.get<1>()->isCompleted()) {
+						if (!a.get<0>()->end_NewData(a.get<1>())) {
+							close(a.get<2>());
+							std::lock_guard<std::mutex> g(lock);
+							backgroundClients.erase(a.get<2>());
+						}
+						return true;
+					}
+					return false;
+				}
+				catch (...) {
+					close(a.get<2>());
+					std::lock_guard<std::mutex> g(lock);
+					backgroundClients.erase(a.get<2>());
+					return true;
+				}
+			}), asyncs.end());
+		lock.lock();
+	}
+	Logger()->messagebf(LOG_DEBUG, "%s: Cleaning up", __PRETTY_FUNCTION__);
+	BOOST_FOREACH(const auto & a, asyncs) {
+		try {
+			a.get<0>()->end_NewData(a.get<1>());
+		}
+		catch (...) {
+		}
+	}
+	backgroundThread = NULL;
+	Logger()->messagebf(LOG_DEBUG, "%s: Unlocking", __PRETTY_FUNCTION__);
+	lock.unlock();
+}
+
+bool
+Tuner::crc32(const P2PVR::Data & buf)
+{
+	boost::crc_optimal<32, 0x0, 0xFFFFFFFF, 0x0, true, false> crc;
+	crc.process_bytes(&buf.front(), buf.size());
+	return crc.checksum() == 0;
+}
+