Skip to content
Projects
Groups
Snippets
Help
This project
Loading...
Sign in / Register
Toggle navigation
U
uniset2
Project
Project
Details
Activity
Cycle Analytics
Repository
Repository
Files
Commits
Branches
Tags
Contributors
Graph
Compare
Charts
Issues
1
Issues
1
List
Board
Labels
Milestones
Merge Requests
0
Merge Requests
0
CI / CD
CI / CD
Pipelines
Jobs
Schedules
Charts
Wiki
Wiki
Members
Members
Collapse sidebar
Close sidebar
Activity
Graph
Charts
Create a new issue
Jobs
Commits
Issue Boards
Open sidebar
UniSet project repositories
uniset2
Commits
1bf3a239
Commit
1bf3a239
authored
Feb 27, 2016
by
Pavel Vainerman
Browse files
Options
Browse Files
Download
Email Patches
Plain Diff
(UNetSender): сделал так, чтобы процесс не вылетал если не удалось создать
соединение, а периодически пытался его создать..
parent
333cf17c
Show whitespace changes
Inline
Side-by-side
Showing
4 changed files
with
88 additions
and
41 deletions
+88
-41
UNetExchange.cc
extensions/UNetUDP/UNetExchange.cc
+4
-2
UNetReceiver.cc
extensions/UNetUDP/UNetReceiver.cc
+1
-1
UNetSender.cc
extensions/UNetUDP/UNetSender.cc
+64
-37
UNetSender.h
extensions/UNetUDP/UNetSender.h
+19
-1
No files found.
extensions/UNetUDP/UNetExchange.cc
View file @
1bf3a239
...
...
@@ -160,8 +160,9 @@ UNetExchange::UNetExchange(UniSetTypes::ObjectId objId, UniSetTypes::ObjectId sh
}
unetinfo
<<
myname
<<
"(init): init sender.. my node "
<<
n_it
.
getProp
(
"name"
)
<<
endl
;
sender
=
make_shared
<
UNetSender
>
(
h
,
p
,
shm
,
s_field
,
s_fvalue
,
prefix
);
sender
=
make_shared
<
UNetSender
>
(
h
,
p
,
shm
,
false
,
s_field
,
s_fvalue
,
prefix
);
sender
->
setSendPause
(
sendpause
);
sender
->
setCheckConnectionPause
(
checkConnectionPause
);
loga
->
add
(
sender
->
getLog
());
try
...
...
@@ -170,8 +171,9 @@ UNetExchange::UNetExchange(UniSetTypes::ObjectId objId, UniSetTypes::ObjectId sh
if
(
!
h2
.
empty
()
)
{
unetinfo
<<
myname
<<
"(init): init sender2.. my node "
<<
n_it
.
getProp
(
"name"
)
<<
endl
;
sender2
=
make_shared
<
UNetSender
>
(
h2
,
p2
,
shm
,
s_field
,
s_fvalue
,
prefix
);
sender2
=
make_shared
<
UNetSender
>
(
h2
,
p2
,
shm
,
false
,
s_field
,
s_fvalue
,
prefix
);
sender2
->
setSendPause
(
sendpause
);
sender2
->
setCheckConnectionPause
(
checkConnectionPause
);
loga
->
add
(
sender2
->
getLog
());
}
}
...
...
extensions/UNetUDP/UNetReceiver.cc
View file @
1bf3a239
...
...
@@ -546,7 +546,7 @@ void UNetReceiver::checkConnectionEvent( ev::periodic& tm, int revents )
if
(
!
activated
)
return
;
unetinfo
<<
myname
<<
"(checkConnectionEvent): check connection
..(checkConnectionTime="
<<
checkConnectionTime
<<
")
"
<<
endl
;
unetinfo
<<
myname
<<
"(checkConnectionEvent): check connection
event..
"
<<
endl
;
uniset_mutex_lock
l
(
checkConnMutex
);
if
(
!
createConnection
(
false
)
)
...
...
extensions/UNetUDP/UNetSender.cc
View file @
1bf3a239
...
...
@@ -25,9 +25,9 @@ using namespace std;
using
namespace
UniSetTypes
;
using
namespace
UniSetExtensions
;
// -----------------------------------------------------------------------------
UNetSender
::
UNetSender
(
const
std
::
string
&
s_host
,
const
ost
::
tpport_t
port
,
const
std
::
shared_ptr
<
SMInterface
>&
smi
,
const
std
::
string
&
s_f
,
const
std
::
string
&
s_val
,
const
std
::
string
&
s_prefix
,
size_t
maxDCount
,
size_t
maxACount
)
:
UNetSender
::
UNetSender
(
const
std
::
string
&
s_host
,
const
ost
::
tpport_t
port
,
const
std
::
shared_ptr
<
SMInterface
>&
smi
,
bool
nocheckConnection
,
const
std
::
string
&
s_f
,
const
std
::
string
&
s_val
,
const
std
::
string
&
s_prefix
,
size_t
maxDCount
,
size_t
maxACount
)
:
s_field
(
s_f
),
s_fvalue
(
s_val
),
prefix
(
s_prefix
),
...
...
@@ -64,27 +64,10 @@ UNetSender::UNetSender( const std::string& s_host, const ost::tpport_t port, con
unetinfo
<<
"(UNetSender): UDP set to "
<<
s_host
<<
":"
<<
port
<<
endl
;
ost
::
Thread
::
setException
(
ost
::
Thread
::
throwException
);
try
{
addr
=
s_host
.
c_str
();
udp
=
make_shared
<
ost
::
UDPBroadcast
>
(
addr
,
port
);
}
catch
(
const
std
::
exception
&
e
)
{
ostringstream
s
;
s
<<
myname
<<
": "
<<
e
.
what
();
unetcrit
<<
s
.
str
()
<<
std
::
endl
;
throw
SystemError
(
s
.
str
());
}
catch
(
...
)
{
ostringstream
s
;
s
<<
myname
<<
": catch..."
;
unetcrit
<<
s
.
str
()
<<
std
::
endl
;
throw
SystemError
(
s
.
str
());
}
ptCheckConnection
.
setTiming
(
10000
);
// default 10 сек
createConnection
(
nocheckConnection
);
s_thr
=
make_shared
<
ThreadCreator
<
UNetSender
>
>
(
this
,
&
UNetSender
::
send
);
...
...
@@ -123,6 +106,40 @@ UNetSender::~UNetSender()
{
}
// -----------------------------------------------------------------------------
bool
UNetSender
::
createConnection
(
bool
throwEx
)
{
ost
::
Thread
::
setException
(
ost
::
Thread
::
throwException
);
unetinfo
<<
myname
<<
"(createConnection): .."
<<
endl
;
try
{
udp
=
make_shared
<
ost
::
UDPBroadcast
>
(
addr
,
port
);
}
catch
(
const
std
::
exception
&
e
)
{
ostringstream
s
;
s
<<
myname
<<
"(createConnection): "
<<
e
.
what
();
unetcrit
<<
s
.
str
()
<<
std
::
endl
;
if
(
throwEx
)
throw
SystemError
(
s
.
str
());
udp
=
nullptr
;
}
catch
(
...
)
{
ostringstream
s
;
s
<<
myname
<<
"(createConnection): catch..."
;
unetcrit
<<
s
.
str
()
<<
std
::
endl
;
if
(
throwEx
)
throw
SystemError
(
s
.
str
());
udp
=
nullptr
;
}
return
(
udp
!=
nullptr
);
}
// -----------------------------------------------------------------------------
void
UNetSender
::
updateFromSM
()
{
auto
it
=
dlist
.
begin
();
...
...
@@ -173,30 +190,40 @@ void UNetSender::updateItem( DMap::iterator& it, long value )
}
}
// -----------------------------------------------------------------------------
void
UNetSender
::
setCheckConnectionPause
(
int
msec
)
{
if
(
msec
>
0
)
ptCheckConnection
.
setTiming
(
msec
);
}
// -----------------------------------------------------------------------------
void
UNetSender
::
send
()
{
dlist
.
resize
(
maxItem
);
unetinfo
<<
myname
<<
"(send): dlist size = "
<<
dlist
.
size
()
<<
endl
;
ncycle
=
0
;
/*
ost::IPV4Broadcast h = s_host.c_str();
try
ptCheckConnection
.
reset
();
while
(
activated
)
{
udp->setPeer(h,port);
}
catch( ost::SockException& e )
if
(
!
udp
)
{
ostringstream s;
s << e.getString() << ": " << e.getSystemErrorString();
unetcrit << myname << "(poll): " << s.str() << endl
;
throw SystemError(s.str())
;
if
(
!
ptCheckConnection
.
checkTime
()
)
{
msleep
(
sendpause
)
;
continue
;
}
*/
ncycle
=
0
;
unetinfo
<<
myname
<<
"(send): check connection event.."
<<
endl
;
while
(
activated
)
if
(
!
createConnection
(
false
)
)
{
ptCheckConnection
.
reset
();
msleep
(
sendpause
);
continue
;
}
}
try
{
if
(
!
shm
->
isLocalwork
()
)
...
...
@@ -274,7 +301,7 @@ void UNetSender::real_send(UniSetUDP::UDPMessage& mypack)
if
(
packetnum
==
0
)
packetnum
=
1
;
if
(
!
udp
->
isPending
(
ost
::
Socket
::
pendingOutput
)
)
if
(
!
udp
||
!
udp
->
isPending
(
ost
::
Socket
::
pendingOutput
)
)
return
;
mypack
.
transport_msg
(
s_msg
);
...
...
extensions/UNetUDP/UNetSender.h
View file @
1bf3a239
...
...
@@ -47,11 +47,24 @@
* ОПТИМИЗАЦИЯ N1: Для оптимизации обработки посылаемых пакетов (на стороне UNetReceiver) сделана следующая логика:
* Номер очередного посылаемого пакета меняется (увеличивается) только, если изменились данные с момента
последней посылки. Для этого по данным каждый раз производится расчёт UNetUDP::makeCRC() и сравнивается с последним..
*
*
* Создание соединения
* ======================================
* Попытка создать соединение производиться сразу в конструкторе, если это не получается,
* то в потоке "посылки", с заданным периодом (checkConnectionTime) идёт попытка создать соединение..
* и так бесконечно, пока не получиться. Это важно для систем, где в момент загрузки программы
* (в момент создания объекта UNetSender) ещё может быть не поднята сеть или какой-то сбой с сетью и требуется
* ожидание (без вылета программы) пока "внешняя система мониторинга" не поднимет сеть).
* Если такая логика не требуется, то можно задать в конструкторе флаг nocheckconnection=true,
* тогда при создании объекта UNetSender, в конструкторе будет
* выкинуто исключение при неудачной попытке создания соединения.
* \warning setCheckConnectionPause(msec) должно быть кратно sendpause!
*/
class
UNetSender
{
public
:
UNetSender
(
const
std
::
string
&
host
,
const
ost
::
tpport_t
port
,
const
std
::
shared_ptr
<
SMInterface
>&
smi
,
UNetSender
(
const
std
::
string
&
host
,
const
ost
::
tpport_t
port
,
const
std
::
shared_ptr
<
SMInterface
>&
smi
,
bool
nocheckConnection
=
false
,
const
std
::
string
&
s_field
=
""
,
const
std
::
string
&
s_fvalue
=
""
,
const
std
::
string
&
prefix
=
"unet"
,
size_t
maxDCount
=
UniSetUDP
::
MaxDCount
,
size_t
maxACount
=
UniSetUDP
::
MaxACount
);
...
...
@@ -106,6 +119,8 @@ class UNetSender
packsendpause
=
msec
;
}
void
setCheckConnectionPause
(
int
msec
);
/*! заказать датчики */
void
askSensors
(
UniversalIO
::
UIOCommand
cmd
);
...
...
@@ -151,6 +166,8 @@ class UNetSender
void
readConfiguration
();
bool
createConnection
(
bool
throwEx
);
private
:
UNetSender
();
...
...
@@ -163,6 +180,7 @@ class UNetSender
timeout_t
sendpause
=
{
150
};
timeout_t
packsendpause
=
{
5
};
std
::
atomic_bool
activated
=
{
false
};
PassiveTimer
ptCheckConnection
;
UniSetTypes
::
uniset_rwmutex
pack_mutex
;
...
...
Write
Preview
Markdown
is supported
0%
Try again
or
attach a new file
Attach a file
Cancel
You are about to add
0
people
to the discussion. Proceed with caution.
Finish editing this message first!
Cancel
Please
register
or
sign in
to comment