<commit_before>#include <sstream>
#include "Exceptions.h"
#include "Extensions.h"
#include "UNetSender.h"
// -----------------------------------------------------------------------------
using namespace std;
using namespace UniSetTypes;
using namespace UniSetExtensions;
// -----------------------------------------------------------------------------
UNetSender::UNetSender( const std::string s_host, const ost::tpport_t port, SMInterface* smi,
						const std::string s_f, const std::string s_val, SharedMemory* ic ):
s_field(s_f),
s_fvalue(s_val),
shm(smi),
s_host(s_host),
sendpause(150),
activated(false),
dlist(100),
maxItem(0),
packetnum(1),
s_thr(0)
{

	{
		ostringstream s;
		s << "(" << s_host << ":" << port << ")";
		myname = s.str();
	}

	// определяем фильтр
//	s_field = conf->getArgParam("--udp-filter-field");
//	s_fvalue = conf->getArgParam("--udp-filter-value");
	dlog[Debug::INFO] << myname << "(init): read filter-field='" << s_field
						<< "' filter-value='" << s_fvalue << "'" << endl;

	if( dlog.debugging(Debug::INFO) )
		dlog[Debug::INFO] << "(UNetSender): UDP set to " << s_host << ":" << port << endl;

	ost::Thread::setException(ost::Thread::throwException);
	try
	{
		addr = s_host.c_str();
		udp = new ost::UDPBroadcast(addr,port);
	}
	catch( std::exception& e )
	{
		ostringstream s;
		s << myname << ": " << e.what();
		dlog[Debug::CRIT] << s.str() << std::endl;
		throw SystemError(s.str());
	}
	catch( ... )
	{
		ostringstream s;
		s << myname << ": catch...";
		dlog[Debug::CRIT] << s.str() << std::endl;
		throw SystemError(s.str());
	}

	s_thr = new ThreadCreator<UNetSender>(this, &UNetSender::send);

	// -------------------------------
	if( shm->isLocalwork() )
	{
		readConfiguration();
		dlist.resize(maxItem);
		dlog[Debug::INFO] << myname << "(init): dlist size = " << dlist.size() << endl;
	}
	else
		ic->addReadItem( sigc::mem_fun(this,&UNetSender::readItem) );


	// выставляем поля, которые не меняются
	mypack.msg.header.nodeID = conf->getLocalNode();
	mypack.msg.header.procID = shm->ID();
}
// -----------------------------------------------------------------------------
UNetSender::~UNetSender()
{
  	delete s_thr;
	delete udp;
	delete shm;
}
// -----------------------------------------------------------------------------
void UNetSender::updateFromSM()
{
	DMap::iterator it=dlist.begin();
	for( ; it!=dlist.end(); ++it )
	{
		long value = 0;

		if( it->iotype == UniversalIO::DigitalInput || it->iotype == UniversalIO::DigitalOutput )
			value = shm->localGetState(it->dit,it->id) ? 1 : 0;
		else if( it->iotype == UniversalIO::AnalogInput || it->iotype == UniversalIO::AnalogOutput )
			value = shm->localGetValue(it->ait,it->id);
		else
		{
			dlog[Debug::CRIT] << myname << "(update): Unknown iotype for sid=" << it->id << endl;
			continue;
		}

		updateItem(it,value);
	}
}
// -----------------------------------------------------------------------------
void UNetSender::updateSensor( UniSetTypes::ObjectId id, long value )
{
//	cerr << myname << ": UPDATE SENSOR id=" << id << " value=" << value << endl;
	DMap::iterator it=dlist.begin();
	for( ; it!=dlist.end(); ++it )
	{
		if( it->id == id )
		{
			updateItem( it, value );
			break;
		}
	}
}
// -----------------------------------------------------------------------------
void UNetSender::updateItem( DMap::iterator& it, long value )
{
	if( it == dlist.end() )
		return;

	if( it->iotype == UniversalIO::DigitalInput || it->iotype == UniversalIO::DigitalOutput )
	{
		UniSetTypes::uniset_mutex_lock l(pack_mutex,100);
		mypack.setDData(it->pack_ind,value);
	}
	else if( it->iotype == UniversalIO::AnalogInput || it->iotype == UniversalIO::AnalogOutput )
	{
		UniSetTypes::uniset_mutex_lock l(pack_mutex,100);
		mypack.setAData(it->pack_ind,value);
	}
}
// -----------------------------------------------------------------------------
void UNetSender::send()
{
	dlist.resize(maxItem);
	dlog[Debug::INFO] << myname << "(send): dlist size = " << dlist.size() << endl;
/*
	ost::IPV4Broadcast h = s_host.c_str();
	try
	{			
		udp->setPeer(h,port);
	}
	catch( ost::SockException& e )
	{
		ostringstream s;
		s << e.getString() << ": " << e.getSystemErrorString();
		dlog[Debug::CRIT] << myname << "(poll): " << s.str() << endl;
		throw SystemError(s.str());
	}
*/
	while( activated )
	{
		try
		{
			if( !shm->isLocalwork() )
				updateFromSM();
			
			real_send();
		}
		catch( ost::SockException& e )
		{
			dlog[Debug::WARN]  << myname << "(send): " << e.getString() << endl;
		}
		catch( UniSetTypes::Exception& ex)
		{
			dlog[Debug::WARN] << myname << "(send): " << ex << std::endl;
		}
		catch( std::exception& e )
		{
			dlog[Debug::WARN] << myname << "(send): " << e.what() << std::endl;
		}
		catch(...)
		{
			dlog[Debug::WARN] << myname << "(send): catch ..." << std::endl;
		}	

		msleep(sendpause);
	}

	dlog[Debug::INFO] << "************* execute FINISH **********" << endl;
}
// -----------------------------------------------------------------------------
void UNetSender::real_send()
{
	UniSetTypes::uniset_mutex_lock l(pack_mutex,300);
	mypack.msg.header.num = packetnum++;

	if( packetnum > UniSetUDP::MaxPacketNum )
		packetnum = 1;

//	cout << "************* send header: " << mypack.msg.header << endl;
//	size_t sz = mypack.byte_size() + sizeof(UniSetUDP::UDPHeader);
	size_t sz =	sizeof(UniSetUDP::UDPMessage);
	if( !udp->isPending(ost::Socket::pendingOutput) )
		return;

	size_t ret = udp->send( (char*)&(mypack.msg),sz);
	if( ret < sz )
		dlog[Debug::CRIT] << myname << "(real_send): FAILED ret=" << ret << " < sizeof=" << sz << endl;
}
// -----------------------------------------------------------------------------
void UNetSender::stop()
{
	activated = false;
//	s_thr->stop();
}
// -----------------------------------------------------------------------------
void UNetSender::start()
{
	if( !activated )<commit_msg>(unet2): исправил ещё один баг<commit_after>#include <sstream>
#include "Exceptions.h"
#include "Extensions.h"
#include "UNetSender.h"
// -----------------------------------------------------------------------------
using namespace std;
using namespace UniSetTypes;
using namespace UniSetExtensions;
// -----------------------------------------------------------------------------
UNetSender::UNetSender( const std::string s_host, const ost::tpport_t port, SMInterface* smi,
						const std::string s_f, const std::string s_val, SharedMemory* ic ):
s_field(s_f),
s_fvalue(s_val),
shm(smi),
s_host(s_host),
sendpause(150),
activated(false),
dlist(100),
maxItem(0),
packetnum(1),
s_thr(0)
{

	{
		ostringstream s;
		s << "(" << s_host << ":" << port << ")";
		myname = s.str();
	}

	// определяем фильтр
//	s_field = conf->getArgParam("--udp-filter-field");
//	s_fvalue = conf->getArgParam("--udp-filter-value");
	dlog[Debug::INFO] << myname << "(init): read filter-field='" << s_field
						<< "' filter-value='" << s_fvalue << "'" << endl;

	if( dlog.debugging(Debug::INFO) )
		dlog[Debug::INFO] << "(UNetSender): UDP set to " << s_host << ":" << port << endl;

	ost::Thread::setException(ost::Thread::throwException);
	try
	{
		addr = s_host.c_str();
		udp = new ost::UDPBroadcast(addr,port);
	}
	catch( std::exception& e )
	{
		ostringstream s;
		s << myname << ": " << e.what();
		dlog[Debug::CRIT] << s.str() << std::endl;
		throw SystemError(s.str());
	}
	catch( ... )
	{
		ostringstream s;
		s << myname << ": catch...";
		dlog[Debug::CRIT] << s.str() << std::endl;
		throw SystemError(s.str());
	}

	s_thr = new ThreadCreator<UNetSender>(this, &UNetSender::send);

	// -------------------------------
	if( shm->isLocalwork() )
	{
		readConfiguration();
		dlist.resize(maxItem);
		dlog[Debug::INFO] << myname << "(init): dlist size = " << dlist.size() << endl;
	}
	else
		ic->addReadItem( sigc::mem_fun(this,&UNetSender::readItem) );


	// выставляем поля, которые не меняются
	mypack.msg.header.nodeID = conf->getLocalNode();
	mypack.msg.header.procID = shm->ID();
}
// -----------------------------------------------------------------------------
UNetSender::~UNetSender()
{
  	delete s_thr;
	delete udp;
	delete shm;
}
// -----------------------------------------------------------------------------
void UNetSender::updateFromSM()
{
	DMap::iterator it=dlist.begin();
	for( ; it!=dlist.end(); ++it )
	{
		long value = 0;

		if( it->iotype == UniversalIO::DigitalInput || it->iotype == UniversalIO::DigitalOutput )
			value = shm->localGetState(it->dit,it->id) ? 1 : 0;
		else if( it->iotype == UniversalIO::AnalogInput || it->iotype == UniversalIO::AnalogOutput )
			value = shm->localGetValue(it->ait,it->id);
		else
		{
			dlog[Debug::CRIT] << myname << "(update): Unknown iotype for sid=" << it->id << endl;
			continue;
		}

		updateItem(it,value);
	}
}
// -----------------------------------------------------------------------------
void UNetSender::updateSensor( UniSetTypes::ObjectId id, long value )
{
//	cerr << myname << ": UPDATE SENSOR id=" << id << " value=" << value << endl;
	DMap::iterator it=dlist.begin();
	for( ; it!=dlist.end(); ++it )
	{
		if( it->id == id )
		{
			updateItem( it, value );
			break;
		}
	}
}
// -----------------------------------------------------------------------------
void UNetSender::updateItem( DMap::iterator& it, long value )
{
	if( it == dlist.end() )
		return;

	if( it->iotype == UniversalIO::DigitalInput || it->iotype == UniversalIO::DigitalOutput )
	{
		UniSetTypes::uniset_mutex_lock l(pack_mutex,100);
		mypack.setDData(it->pack_ind,value);
	}
	else if( it->iotype == UniversalIO::AnalogInput || it->iotype == UniversalIO::AnalogOutput )
	{
		UniSetTypes::uniset_mutex_lock l(pack_mutex,100);
		mypack.setAData(it->pack_ind,value);
	}
}
// -----------------------------------------------------------------------------
void UNetSender::send()
{
	dlist.resize(maxItem);
	dlog[Debug::INFO] << myname << "(send): dlist size = " << dlist.size() << endl;
/*
	ost::IPV4Broadcast h = s_host.c_str();
	try
	{			
		udp->setPeer(h,port);
	}
	catch( ost::SockException& e )
	{
		ostringstream s;
		s << e.getString() << ": " << e.getSystemErrorString();
		dlog[Debug::CRIT] << myname << "(poll): " << s.str() << endl;
		throw SystemError(s.str());
	}
*/
	while( activated )
	{
		try
		{
			if( !shm->isLocalwork() )
				updateFromSM();
			
			real_send();
		}
		catch( ost::SockException& e )
		{
			dlog[Debug::WARN]  << myname << "(send): " << e.getString() << endl;
		}
		catch( UniSetTypes::Exception& ex)
		{
			dlog[Debug::WARN] << myname << "(send): " << ex << std::endl;
		}
		catch( std::exception& e )
		{
			dlog[Debug::WARN] << myname << "(send): " << e.what() << std::endl;
		}
		catch(...)
		{
			dlog[Debug::WARN] << myname << "(send): catch ..." << std::endl;
		}	

		msleep(sendpause);
	}

	dlog[Debug::INFO] << "************* execute FINISH **********" << endl;
}
// -----------------------------------------------------------------------------
void UNetSender::real_send()
{
	UniSetTypes::uniset_mutex_lock l(pack_mutex,300);
	mypack.msg.header.num = packetnum++;

	if( packetnum > UniSetUDP::MaxPacketNum )
		packetnum = 1;

//	cout << "************* send header: " << mypack.msg.header << endl;
//	size_t sz = mypack.byte_size() + sizeof(UniSetUDP::UDPHeader);
	size_t sz =	sizeof(UniSetUDP::DataPacket);
	if( !udp->isPending(ost::Socket::pendingOutput) )
		return;

	size_t ret = udp->send( (char*)&(mypack.msg),sz);
	if( ret < sz )
		dlog[Debug::CRIT] << myname << "(real_send): FAILED ret=" << ret << " < sizeof=" << sz << endl;
}
// -----------------------------------------------------------------------------
void UNetSender::stop()
{
	activated = false;
//	s_thr->stop();
}
// -----------------------------------------------------------------------------
void UNetSender::start()
{
	if( !activated )
{'fraction_non_alphanumeric': 0.32243507875691785, 'fraction_numerical': 0.004001702852277565, 'mean_word_length': 2.2474426320154826, 'pattern_counts': {'":': 12, '<': 319, '<?xml version=': 0, '>': 67, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3196952', 'n_tokens_mistral': 4303, 'n_tokens_neox': 4009, 'n_words': 718}