pip/libs/main/io_devices/piiodevice.cpp

/*
    PIP - Platform Independent Primitives
    Abstract input/output device
    Ivan Pelipenko peri4ko@yandex.ru, Andrey Bychkov work.a.b@yandex.ru

    This program is free software: you can redistribute it and/or modify
    it under the terms of the GNU Lesser General Public License as published by
    the Free Software Foundation, either version 3 of the License, or
    (at your option) any later version.

    This program is distributed in the hope that it will be useful,
    but WITHOUT ANY WARRANTY; without even the implied warranty of
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
    GNU Lesser General Public License for more details.

    You should have received a copy of the GNU Lesser General Public License
    along with this program.  If not, see <http://www.gnu.org/licenses/>.
*/

#include "piiodevice.h"

#include "piconfig.h"
#include "piliterals_bytes.h"
#include "piliterals_string.h"
#include "piliterals_time.h"
#include "pipropertystorage.h"
#include "pitime.h"
#include "pitranslator.h"


//! \class PIIODevice piiodevice.h
//! \~\details
//! \section PIIODevice_sec0 Synopsis
//! This class provide open/close logic, threaded read/write and virtual input/output
//! functions \a read() and \a write(). You should implement pure virtual
//! function \a openDevice() in your subclass.
//!
//! \section PIIODevice_sec1 Open and close
//! PIIODevice have boolean variable indicated open status. Returns of functions
//! \a openDevice() and \a closeDevice() change this variable.
//!
//! \section PIIODevice_sec2 Threaded read
//! PIIODevice based on PIThread, so it`s overload \a run() to exec \a read()
//! in background thread. If read is successful virtual function \a threadedRead()
//! is executed. Default implementation of this function execute external static
//! function set by \a setThreadedReadSlot() with data set by \a setThreadedReadData().
//! Extrenal static function should have format \n
//! bool func_name(void * Threaded_read_data, uchar * readed_data, int readed_size)\n
//! Threaded read starts with function \a startThreadedRead().
//!
//! \section PIIODevice_sec3 Threaded write
//! PIIODevice aggregate another PIThread to perform a threaded write by function
//! \a writeThreaded(). This function add task to internal queue and return
//! queue entry ID. You should start write thread by function \a startThreadedWrite.
//! On successful write event \a threadedWriteEvent is raised with two arguments -
//! task ID and written bytes count.
//!
//! \section PIIODevice_sec4 Internal buffer
//! PIIODevice have internal buffer for threaded read, and \a threadedRead() function
//! receive pointer to this buffer in first argument. You can adjust size of this buffer
//! by function \a setThreadedReadBufferSize() \n
//! Default size of this buffer is 4096 bytes.
//!
//! \section PIIODevice_sec5 Reopen
//! When threaded read is begin its call \a open() if device is closed. While threaded
//! read running PIIODevice check if device opened every read and if not call \a open()
//! every reopen timeout if reopen enabled. Reopen timeout is set by \a setReopenTimeout(),
//! reopen enable is set by \a setReopenEnabled().
//!
//! \section PIIODevice_sec6 Configuration
//! This is virtual function \a configureDevice() which executes when \a configure()
//! executes. This function takes two arguments: "e_main" and "e_parent" as void*. There
//! are pointers to PIConfig::Entry entries of section "section" and their parent. If
//! there is no parent "e_parent" = 0. Function \a configure() set three parameters of
//! device: "reopenEnabled", "reopenTimeout" and "threadedReadBufferSize", then execute
//! function \a configureDevice().
//! \n Each ancestor of %PIIODevice reimlements \a configureDevice() function to be able
//! to be confured from configuration file. This parameters described at section
//! "Configurable parameters" in the class reference. \n Usage example:
//! \snippet piiodevice.cpp configure
//! Implementation example:
//! \snippet piiodevice.cpp configureDevice
//!
//! \section PIIODevice_sec7 Creating devices by unambiguous string
//! There are some virtual functions to describe child class without its declaration.
//! \n \a constructFullPath() should returns full unambiguous string, contains prefix and all device parameters
//! \n \a configureFromFullPath() provide configuring device from full unambiguous string without prefix and "://"
//! \n Macro PIIODEVICE should be used instead of PIOBJECT
//! \n Macro REGISTER_DEVICE should be used after declaration of class, i.e. at the last line of *.h file
//! \n \n If custom I/O device corresponds there rules, it can be returned by function \a createFromFullPath().
//! \n Each PIP I/O device has custom unambiguous string description:
//! * PIFile: "file://<path>"
//! * PIBinaryLog: "binlog://<logDir>[:<filePrefix>][:<defaultID>]"
//! * PISerial: "ser://<device>:<speed(50|...|115200)>[:<dataBitsCount(6|7|8)>][:<parity(N|E|O)>][:<stopBits(1|2)>]"
//! * PIEthernet: UDP "eth://UDP:<readIP>:<readPort>:<sendIP>:<sendPort>[:<multicast(mcast:<ip>)>]"
//! * PIEthernet: TCP "eth://TCP:<IP>:<Port>"
//! * PIUSB: "usb://<vid>:<pid>[:<deviceNumber>][:<readEndpointNumber>][:<writeEndpointNumber>]"
//! \n \n Examples:
//! * PIFile: "file://../text.txt"
//! * PIBinaryLog: "binlog://../logs/:mylog_:1"
//! * PISerial: "ser:///dev/ttyUSB0:9600:8:N:1", equivalent "ser:///dev/ttyUSB0:9600"
//! * PIEthernet: "eth://TCP:127.0.0.1:16666", "eth://UDP:192.168.0.5:16666:192.168.0.6:16667:mcast:234.0.2.1:mcast:234.0.2.2"
//! * PIUSB: "usb://0bb4:0c86:1:1:2"
//! \n \n
//! So, custom I/O device can be created with next call:
//! \code{cpp}
//! // creatring devices
//! PISerial * ser = (PISerial * )PIIODevice::createFromFullPath("ser://COM1:115200");
//! PIEthernet * eth = (PIEthernet * )PIIODevice::createFromFullPath("eth://UDP:127.0.0.1:4001:127.0.0.1:4002");
//! // examine devices
//! piCout << ser << ser->properties();
//! piCout << eth << eth->properties();
//! \endcode
//!
//! \section PIIODevice_ex0 Example
//! \snippet piiodevice.cpp 0
//!


PIMutex PIIODevice::nfp_mutex;
PIMap<PIString, PIString> PIIODevice::nfp_cache;


PIIODevice::PIIODevice(): PIObject() {
	mode_ = ReadOnly;
	_init();
	setPath(PIString());
}


PIIODevice::PIIODevice(const PIString & path, PIIODevice::DeviceMode mode): PIObject() {
	mode_ = mode;
	_init();
	setPath(path);
}


PIIODevice::~PIIODevice() {
	destroying = true;
	stopAndWait();
}


void PIIODevice::setOptions(PIIODevice::DeviceOptions o) {
	options_ = o;
	optionsChanged();
}


bool PIIODevice::setOption(PIIODevice::DeviceOption o, bool yes) {
	bool ret = isOptionSet(o);
	options_.setFlag(o, yes);
	optionsChanged();
	return ret;
}


void PIIODevice::setReopenEnabled(bool yes) {
	setProperty("reopenEnabled", yes);
	reopen_enabled = yes;
}


void PIIODevice::setReopenTimeout(PISystemTime timeout) {
	setProperty("reopenTimeout", timeout);
	reopen_timeout = timeout;
}


//! \~\details
//! \~english
//! Set external static function of threaded read that will be executed
//! at every successful threaded read. Function should have format
//! "bool func(void * data, uchar * readed, int size)"
//! \~russian
//! Устанавливает внешний статический метод, который будет вызван
//! после каждого успешного потокового чтения. Метод должен быть
//! в формате "bool func(void * data, uchar * readed, int size)"
void PIIODevice::setThreadedReadSlot(ReadRetFunc func) {
	func_read = func;
}


//! \~\details
//! \~english
//! Default size is 4096 bytes. If your device can read at single read
//! more than 4096 bytes you should use this function to adjust buffer size
//! \~russian
//! По умолчанию 4096 байт. Если устройство за одно чтение может читать
//! более 4096 байт, необходимо использовать этот метод для установки
//! нужного размера буфера
void PIIODevice::setThreadedReadBufferSize(int new_size) {
	threaded_read_buffer_size = new_size;
	threadedReadBufferSizeChanged();
}


bool PIIODevice::isThreadedRead() const {
	return read_thread.isRunning();
}


void PIIODevice::startThreadedRead() {
	if (!read_thread.isRunning()) {
		buffer_tr.resize(threaded_read_buffer_size);
		read_thread.start();
	}
}


void PIIODevice::startThreadedRead(ReadRetFunc func) {
	func_read = func;
	startThreadedRead();
}


void PIIODevice::stopThreadedRead() {
	if (!isThreadedRead()) return;
#ifdef MICRO_PIP
	read_thread.stop();
#else
	read_thread.stop();
	if (!destroying) {
		interrupt();
	} else {
		piCoutObj << "Error: Device is running after destructor!"_tr("PIIODevice");
	}
#endif
}


void PIIODevice::terminateThreadedRead() {
	read_thread.terminate();
}


bool PIIODevice::waitThreadedReadFinished(PISystemTime timeout) {
	PITimeMeasurer tm, tm_intr;
	while (read_thread.isRunning()) {
		if (timeout.isNotNull()) {
			if (tm.elapsed() > timeout) return false;
		}
		if (tm_intr.elapsed() > 100_ms) {
			tm_intr.reset();
			interrupt();
		}
		piMinSleep();
	}
	return true;
}


bool PIIODevice::isThreadedWrite() const {
	return write_thread.isRunning();
}


void PIIODevice::startThreadedWrite() {
	if (!write_thread.isRunning()) write_thread.startOnce();
}


void PIIODevice::stopThreadedWrite() {
	if (!write_thread.isRunning()) return;
	write_thread.stop();
}


void PIIODevice::terminateThreadedWrite() {
	write_thread.terminate();
}


bool PIIODevice::waitThreadedWriteFinished(PISystemTime timeout) {
	return write_thread.waitForFinish(timeout);
}


void PIIODevice::clearThreadedWriteQueue() {
	write_thread.lock();
	write_queue.clear();
	write_thread.unlock();
}


void PIIODevice::start() {
	startThreadedRead();
	startThreadedWrite();
}


void PIIODevice::stop() {
	stopThreadedRead();
	stopThreadedWrite();
}


void PIIODevice::stopAndWait(PISystemTime timeout) {
	stop();
	waitThreadedReadFinished(timeout);
	waitThreadedWriteFinished(timeout);
}


ssize_t PIIODevice::read(void * read_to, ssize_t max_size) {
	ssize_t ret = readDevice(read_to, max_size);
	return ret;
}


ssize_t PIIODevice::read(PIMemoryBlock mb) {
	return read(mb.data(), mb.size());
}


PIByteArray PIIODevice::read(ssize_t max_size) {
	if (max_size <= 0) return PIByteArray();
	buffer_in.resize(max_size);
	ssize_t ret = readDevice(buffer_in.data(), max_size);
	if (ret < 0) return PIByteArray();
	return buffer_in.resized(ret);
}


ssize_t PIIODevice::write(const void * data, ssize_t max_size) {
	if (max_size <= 0) return 0;
	return writeDevice(data, max_size);
}


void PIIODevice::_init() {
	opened_ = false;
	setOptions(0);
	setReopenEnabled(true);
	setReopenTimeout(1_s);
#ifdef MICRO_PIP
	threaded_read_buffer_size = 512;
#else
	threaded_read_buffer_size = 4_KiB;
#endif
	read_thread.setName("_S.PIIODev.read");
	write_thread.setName("_S.PIIODev.write");
	CONNECT(void, &write_thread, started, this, write_func);
	CONNECTL(&read_thread, started, [this]() {
		if (!isOpened()) open();
	});
	read_thread.setSlot([this](void *) { read_func(); });
}


void PIIODevice::write_func() {
	while (!write_thread.isStopping()) {
		while (!write_queue.isEmpty()) {
			if (write_thread.isStopping()) return;
			write_thread.lock();
			PIPair<PIByteArray, ullong> item(write_queue.dequeue());
			write_thread.unlock();
			int ret = write(item.first);
			threadedWriteEvent(item.second, ret);
		}
		piMinSleep();
	}
}


PIIODevice * PIIODevice::newDeviceByPrefix(const char * prefix) {
	if (!prefix) return nullptr;
	auto fi = fabrics().value(prefix);
	if (fi.fabricator) return fi.fabricator();
	return nullptr;
}


void PIIODevice::read_func() {
	if (!isReadable()) {
		read_thread.stop();
		return;
	}
	if (!isOpened()) {
		piMSleep(10);
		bool ok = false;
		if (reopen_enabled) {
			if (reopen_tm.elapsed() >= reopen_timeout) {
				reopen_tm.reset();
				ok = open();
			}
		}
		if (!ok || read_thread.isStopping()) return;
	}
	ssize_t readed_ = read(buffer_tr.data(), buffer_tr.size_s());
	if (read_thread.isStopping()) return;
	if (readed_ <= 0) {
		piMSleep(10);
		// cout << readed_ << ", " << errno << ", " << errorString() << endl;
		return;
	}
	// piCoutObj << "readed" << readed_;// << ", " << errno << ", " << errorString();
	threadedRead(buffer_tr.data(), readed_);
	threadedReadEvent(buffer_tr.data(), readed_);
}


PIByteArray PIIODevice::readForTime(PISystemTime timeout) {
	PIByteArray str;
	if (timeout.isNull()) return str;
	ssize_t ret;
	uchar * td  = new uchar[threaded_read_buffer_size];
	bool was_br = setOption(BlockingRead, false);
	tm.reset();
	while (tm.elapsed() < timeout) {
		ret = read(td, threaded_read_buffer_size);
		if (ret <= 0)
			piMinSleep();
		else
			str.append(td, ret);
	}
	setOption(BlockingRead, was_br);
	delete[] td;
	return str;
}


ullong PIIODevice::writeThreaded(const PIByteArray & data) {
	write_thread.lock();
	write_queue.enqueue(PIPair<PIByteArray, ullong>(data, tri));
	++tri;
	write_thread.unlock();
	return tri - 1;
}


bool PIIODevice::open() {
	buffer_tr.resize(threaded_read_buffer_size);
	opened_ = openDevice();
	if (opened_) opened();
	return opened_;
}


bool PIIODevice::open(const PIString & _path) {
	setPath(_path);
	return open();
}


bool PIIODevice::open(DeviceMode _mode) {
	mode_ = _mode;
	return open();
}


bool PIIODevice::open(const PIString & _path, DeviceMode _mode) {
	setPath(_path);
	mode_ = _mode;
	return open();
}


bool PIIODevice::close() {
	opened_ = !closeDevice();
	if (!opened_) closed();
	return !opened_;
}


ssize_t PIIODevice::write(PIByteArray data) {
	return writeDevice(data.data(), data.size_s());
}


bool PIIODevice::configure(const PIString & config_file, const PIString & section, bool parent_section) {
	PIConfig conf(config_file, PIIODevice::ReadOnly);
	if (!conf.isOpened()) return false;
	bool ex = true;
	PIConfig::Entry em;
	if (section.isEmpty())
		em = conf.rootEntry();
	else
		em = conf.getValue(section, PIString(), &ex);
	if (!ex) return false;
	PIConfig::Entry * ep = 0;
	if (parent_section) ep = em.parent();
	if (ep != 0) {
		setReopenEnabled(ep->getValue("reopenEnabled", isReopenEnabled(), &ex).toBool());
		if (!ex) setReopenEnabled(em.getValue("reopenEnabled", isReopenEnabled()).toBool());
		setReopenTimeout(PISystemTime::fromMilliseconds(ep->getValue("reopenTimeout", reopenTimeout().toMilliseconds(), &ex).toInt()));
		if (!ex) setReopenTimeout(PISystemTime::fromMilliseconds(em.getValue("reopenTimeout", reopenTimeout().toMilliseconds()).toInt()));
		setThreadedReadBufferSize(ep->getValue("threadedReadBufferSize", int(threaded_read_buffer_size), &ex).toInt());
		if (!ex) setThreadedReadBufferSize(em.getValue("threadedReadBufferSize", int(threaded_read_buffer_size)).toInt());
	} else {
		setReopenEnabled(em.getValue("reopenEnabled", isReopenEnabled()).toBool());
		setReopenTimeout(PISystemTime::fromMilliseconds(em.getValue("reopenTimeout", reopenTimeout().toMilliseconds()).toInt()));
		setThreadedReadBufferSize(em.getValue("threadedReadBufferSize", int(threaded_read_buffer_size)).toInt());
	}
	return configureDevice(&em, ep);
}


PIString PIIODevice::constructFullPath() const {
	return fullPathPrefix().toString() + "://"_a + constructFullPathDevice() + fullPathOptions();
}


void PIIODevice::configureFromFullPath(const PIString & full_path) {
	PIString fp;
	DeviceMode dm    = ReadWrite;
	DeviceOptions op = 0;
	splitFullPath(full_path, &fp, &dm, &op);
	setMode(dm);
	setOptions(op);
	configureFromFullPathDevice(fp);
}


PIVariantTypes::IODevice PIIODevice::constructVariant() const {
	PIVariantTypes::IODevice ret;
	ret.prefix  = fullPathPrefix();
	ret.mode    = mode();
	ret.options = options();
	ret.set(constructVariantDevice());
	return ret;
}


void PIIODevice::configureFromVariant(const PIVariantTypes::IODevice & d) {
	setMode((DeviceMode)d.mode);
	setOptions((DeviceOptions)d.options);
	configureFromVariantDevice(d.get());
}


void PIIODevice::splitFullPath(PIString fpwm, PIString * full_path, DeviceMode * mode, DeviceOptions * opts) {
	fpwm.trim();
	int dm           = 0;
	DeviceOptions op = 0;
	if (fpwm.find('(') > 0 && fpwm.find(')') > 0) {
		PIString dms(fpwm.right(fpwm.length() - fpwm.findLast('(')).takeRange('(', ')').trim().toLowerCase().removeAll(' '));
		PIStringList opts(dms.split(","));
		for (const auto & o: opts) {
			// piCout << dms;
			if (o == "r"_a || o == "ro"_a || o == "read"_a || o == "readonly"_a) dm |= ReadOnly;
			if (o == "w"_a || o == "wo"_a || o == "write"_a || o == "writeonly"_a) dm |= WriteOnly;
			if (o == "br"_a || o == "blockr"_a || o == "blockread"_a || o == "blockingread"_a) op |= BlockingRead;
			if (o == "bw"_a || o == "blockw"_a || o == "blockwrite"_a || o == "blockingwrite"_a) op |= BlockingWrite;
			if (o == "brw"_a || o == "bwr"_a || o == "blockrw"_a || o == "blockwr"_a || o == "blockreadrite"_a ||
				o == "blockingreadwrite"_a)
				op |= BlockingRead | BlockingWrite;
		}
		fpwm.cutRight(fpwm.length() - fpwm.findLast('(')).trim();
	}
	if (dm == 0) dm = ReadWrite;
	if (full_path) *full_path = fpwm;
	if (mode) *mode = (DeviceMode)dm;
	if (opts) *opts = op;
}


PIStringList PIIODevice::availablePrefixes() {
	PIStringList ret;
	for (const auto & i: fabrics())
		ret << i.second.prefix.toString();
	return ret;
}


PIStringList PIIODevice::availableClasses() {
	PIStringList ret;
	for (const auto & i: fabrics())
		ret << i.second.classname.toString();
	return ret;
}


void PIIODevice::registerDevice(PIConstChars prefix, PIConstChars classname, PIIODevice * (*fabric)()) {
	// printf("registerDevice %s %s %d\n", prefix.data(), classname.data(), fabrics().size());
	if (prefix.isEmpty()) return;
	if (fabrics().contains(prefix)) return;
	FabricInfo fi;
	fi.prefix         = prefix;
	fi.classname      = classname;
	fi.fabricator     = fabric;
	fabrics()[prefix] = fi;
}


PIString PIIODevice::fullPathOptions() const {
	if (mode_ == ReadWrite && options_ == 0) return PIString();
	PIString ret(" ("_a);
	bool f = true;
	if (mode_ == ReadOnly) {
		if (!f) ret += ","_a;
		f = false;
		ret += "ro"_a;
	}
	if (mode_ == WriteOnly) {
		if (!f) ret += ","_a;
		f = false;
		ret += "wo"_a;
	}
	if (options_[BlockingRead]) {
		if (!f) ret += ","_a;
		f = false;
		ret += "br"_a;
	}
	if (options_[BlockingWrite]) {
		if (!f) ret += ","_a;
		f = false;
		ret += "bw"_a;
	}
	return ret + ")"_a;
}


//! \~\details
//! \~english
//! To function \a configureFromFullPath() "full_path" passed without \a fullPathPrefix() and "://".
//! See \ref PIIODevice_sec7
//! \~russian
//! В метод \a configureFromFullPath() "full_path" передается без \a fullPathPrefix() и "://".
//! См. \ref PIIODevice_sec7
PIIODevice * PIIODevice::createFromFullPath(const PIString & full_path) {
	PIString fp     = full_path.trimmed();
	PIString prefix = fp.left(fp.find(":"));
	PIIODevice * nd = newDeviceByPrefix(prefix.dataAscii());
	if (!nd) return nullptr;
	nd->configureFromFullPath(fp.mid(prefix.length() + 3));
	cacheFullPath(fp, nd);
	return nd;
}


PIIODevice * PIIODevice::createFromVariant(const PIVariantTypes::IODevice & d) {
	PIIODevice * nd = newDeviceByPrefix(d.prefix.dataAscii());
	if (!nd) return nullptr;
	nd->configureFromVariant(d);
	return nd;
}


PIString PIIODevice::normalizeFullPath(const PIString & full_path) {
	nfp_mutex.lock();
	PIString ret = nfp_cache.value(full_path);
	if (!ret.isEmpty()) {
		nfp_mutex.unlock();
		return ret;
	}
	nfp_mutex.unlock();
	PIIODevice * d = createFromFullPath(full_path);
	// piCout << "normalizeFullPath" << d;
	if (d == 0) return PIString();
	ret = d->constructFullPath();
	delete d;
	return ret;
}


void PIIODevice::cacheFullPath(const PIString & full_path, const PIIODevice * d) {
	PIMutexLocker nfp_ml(nfp_mutex);
	nfp_cache[full_path] = d->constructFullPath();
}


PIMap<PIConstChars, PIIODevice::FabricInfo> & PIIODevice::fabrics() {
	static PIMap<PIConstChars, FabricInfo> ret;
	return ret;
}


bool PIIODevice::threadedRead(const uchar * readed, ssize_t size) {
	// piCout << "iodevice threaded read";
	if (func_read) return func_read(readed, size, ret_data_);
	return true;
}


PIPropertyStorage PIIODevice::constructVariantDevice() const {
	PIPropertyStorage ret;
	ret.addProperty("path", path());
	return ret;
}


void PIIODevice::configureFromVariantDevice(const PIPropertyStorage & d) {
	setPath(d.propertyValueByName("path").toString());
}