1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
|
// $Id$
#include "orbsvcs/Log_Macros.h"
#include "orbsvcs/PortableGroup/UIPMC_Transport.h"
#include "orbsvcs/PortableGroup/miopconf.h"
#include "orbsvcs/PortableGroup/UIPMC_Connection_Handler.h"
#include "orbsvcs/PortableGroup/UIPMC_Message_Block_Data_Iterator.h"
#include "orbsvcs/PortableGroup/UIPMC_Wait_Never.h"
#include "orbsvcs/PortableGroup/miop_resource.h"
#include "tao/ORB_Core.h"
#include "tao/debug.h"
#include "tao/GIOP_Message_Base.h"
#include "ace/UUID.h"
TAO_BEGIN_VERSIONED_NAMESPACE_DECL
TAO_UIPMC_Transport::TAO_UIPMC_Transport (
TAO_UIPMC_Connection_Handler *handler,
TAO_ORB_Core *orb_core
)
: TAO_Transport (IOP::TAG_UIPMC, orb_core)
, connection_handler_ (handler)
, total_bytes_outstanding_ (0u)
, time_last_sent_ (ACE_Time_Value::zero)
{
// Replace the default wait strategy with our own
// since we don't support waiting on anything.
delete this->ws_;
ACE_NEW (this->ws_, TAO_UIPMC_Wait_Never (this));
ACE_Utils::UUID uuid;
ACE_Utils::UUID_GENERATOR::instance ()->generate_UUID (uuid);
// This ID is "globally" unique.
this->uuid_hash_ = uuid.to_string ()->hash ();
}
TAO_UIPMC_Transport::~TAO_UIPMC_Transport (void)
{
}
ACE_Event_Handler *
TAO_UIPMC_Transport::event_handler_i (void)
{
return this->connection_handler_;
}
TAO_Connection_Handler *
TAO_UIPMC_Transport::connection_handler_i (void)
{
return this->connection_handler_;
}
bool
TAO_UIPMC_Transport::write_unique_id (
TAO_OutputCDR &miop_hdr) const
{
// This is unique within single machine.
unsigned long pid = static_cast<unsigned long> (ACE_OS::getpid ());
// This is unique within single process.
static ACE_Atomic_Op<TAO_SYNCH_MUTEX, unsigned long> cnt = 0;
unsigned long const id = ++cnt;
CORBA::Octet unique_id[MIOP_DEFAULT_ID_LENGTH];
#define UIPMC_OCTET(value) static_cast<CORBA::Octet> ((value) & 0xff)
// Use the lowest 4 bytes. In case of IPv6 it gives more diversity.
unique_id[0] = UIPMC_OCTET (this->uuid_hash_);
unique_id[1] = UIPMC_OCTET (this->uuid_hash_ >> 8);
unique_id[2] = UIPMC_OCTET (this->uuid_hash_ >> 16);
unique_id[3] = UIPMC_OCTET (this->uuid_hash_ >> 24);
unique_id[4] = UIPMC_OCTET (pid);
unique_id[5] = UIPMC_OCTET (pid >> 8);
unique_id[6] = UIPMC_OCTET (pid >> 16);
unique_id[7] = UIPMC_OCTET (pid >> 24);
unique_id[8] = UIPMC_OCTET (id);
unique_id[9] = UIPMC_OCTET (id >> 8);
unique_id[10] = UIPMC_OCTET (id >> 16);
unique_id[11] = UIPMC_OCTET (id >> 24);
miop_hdr.write_ulong (MIOP_DEFAULT_ID_LENGTH);
miop_hdr.write_octet_array (unique_id, MIOP_DEFAULT_ID_LENGTH);
return miop_hdr.good_bit ();
}
void
TAO_UIPMC_Transport::throttle_send_rate (
ACE_UINT64 const max_fragment_rate,
u_long const max_fragment_size,
u_long const this_fragment_size)
{
ACE_Time_Value const now = ACE_OS::gettimeofday ();
if (this->total_bytes_outstanding_)
{
// How much of the previously sent data have we had time to send?
ACE_Time_Value const how_long = now - this->time_last_sent_;
ACE_UINT64 how_long_in_micro_seconds = 0u;
how_long.to_usec (how_long_in_micro_seconds);
ACE_UINT64 const octets_processed =
how_long_in_micro_seconds * max_fragment_size / max_fragment_rate;
if (this->total_bytes_outstanding_ <= octets_processed)
{
if (2 <= TAO_debug_level)
ORBSVCS_DEBUG ((LM_DEBUG,
ACE_TEXT ("TAO (%P|%t) - UIPMC_Transport[%d]::")
ACE_TEXT ("throttle_send_rate, Previous data ")
ACE_TEXT ("(%u bytes) has cleared ")
ACE_TEXT ("(could have sent %Q bytes over ")
ACE_TEXT ("the last %Q uSecs)\n"),
this->id (),
this->total_bytes_outstanding_,
octets_processed,
how_long_in_micro_seconds));
this->total_bytes_outstanding_ = 0u;
}
else
{
if (2 <= TAO_debug_level)
ORBSVCS_DEBUG ((LM_DEBUG,
ACE_TEXT ("TAO (%P|%t) - UIPMC_Transport[%d]::")
ACE_TEXT ("throttle_send_rate, Previous data ")
ACE_TEXT ("(%u bytes) has reduced by %Q bytes ")
ACE_TEXT ("over the last %Q uSecs\n"),
this->id (),
this->total_bytes_outstanding_,
octets_processed,
how_long_in_micro_seconds));
this->total_bytes_outstanding_ -=
static_cast<u_long> (octets_processed);
}
}
this->time_last_sent_ = now;
if (this->total_bytes_outstanding_)
{
// Is this fragment enough to exceed the capacity of the client
// and/or server system socket buffers?
u_long const
new_total_send = this->total_bytes_outstanding_ + this_fragment_size,
hwm = this->connection_handler_->send_hi_water_mark ();
if (new_total_send > hwm)
{
// Definatly need to throttle back and delay before sending
// this new packet. How much extra data are we trying to fit
// into the various socket buffers?
u_long bytes_over_by = new_total_send - hwm;
// We need to wait until existing "bytes_over_by" data has been
// transmitted/processed; however since we only have
// "total_bytes_outstanding" actualy in progress at the moment,
// this is the maximum amount we can wait for. (This can happen
// if the High Water Mark is set less than the "max_fragment_size"
// and means we can initialy exceed the HWM; will simply wait
// longer before the next fragment, after this current one.)
if (bytes_over_by > this->total_bytes_outstanding_)
bytes_over_by = this->total_bytes_outstanding_;
// How much time does it take for "bytes_over_by" data to be
// processed?
ACE_UINT64 const delay_in_micro_seconds=
bytes_over_by * max_fragment_rate / max_fragment_size;
ACE_Time_Value const delay (
static_cast <time_t> (delay_in_micro_seconds / ACE_ONE_SECOND_IN_USECS),
static_cast <suseconds_t> (delay_in_micro_seconds % ACE_ONE_SECOND_IN_USECS));
if (TAO_debug_level)
ORBSVCS_DEBUG ((LM_DEBUG,
ACE_TEXT ("TAO (%P|%t) - UIPMC_Transport[%d]::")
ACE_TEXT ("throttle_send_rate, SendHighWaterMark ")
ACE_TEXT ("(%u) exceeded by %u bytes, delaying ")
ACE_TEXT ("for %Q uSecs\n"),
this->id (),
hwm,
bytes_over_by,
delay_in_micro_seconds));
ACE_OS::sleep (delay);
}
}
}
ssize_t
TAO_UIPMC_Transport::send (
iovec *iov,
int iovcnt,
size_t &bytes_transferred,
const ACE_Time_Value *)
{
bytes_transferred = 0u;
// Calculate the total number of bytes being sent.
u_long bytes_to_send = 0u;
for (int i = 0; i < iovcnt; ++i)
bytes_to_send += iov[i].iov_len;
if (!bytes_to_send)
return 0; // Nothing to send, we are done.
// Determine the number of MIOP packets that we have to send (add one
// if "bytes_to_send" is not a multiple of "max_fragment_payload").
TAO_MIOP_Resource_Factory *const factory =
ACE_Dynamic_Service<TAO_MIOP_Resource_Factory>::instance (
this->orb_core_->configuration(),
ACE_TEXT ("MIOP_Resource_Factory"));
u_long const
max_fragment_size= factory->max_fragment_size (),
max_fragment_payload= max_fragment_size - MIOP_DEFAULT_HEADER_SIZE,
number_of_packets_required =
bytes_to_send / max_fragment_payload +
!!(bytes_to_send % max_fragment_payload);
if (factory->max_fragments () && // Any value except Zero (Unlimited)
factory->max_fragments () < number_of_packets_required)
{
// Pretend it was sent ok, but we are dropping this message,
// it is too large to send.
bytes_transferred = bytes_to_send;
if (TAO_debug_level)
{
ORBSVCS_DEBUG ((LM_DEBUG,
ACE_TEXT ("TAO (%P|%t) - UIPMC_Transport[%d]::send, ")
ACE_TEXT ("Did not send MIOP message of size %u ")
ACE_TEXT ("(it was too large, needing %u fragments).\n")
ACE_TEXT ("TAO (%P|%t) - UIPMC_Transport[%d]::send, ")
ACE_TEXT ("You maybe able to increase ")
ACE_TEXT ("MIOP_Resource_Factory -ORBMaxFragments %u"),
this->id (),
bytes_to_send,
number_of_packets_required,
this->id (),
factory->max_fragments ()));
if (max_fragment_size < MIOP_MAX_DGRAM_SIZE)
ORBSVCS_DEBUG ((LM_DEBUG,
ACE_TEXT (" or -ORBMaxFragmentSize %u\n"),
max_fragment_size));
else
ORBSVCS_DEBUG ((LM_DEBUG, ACE_TEXT ("\n")));
}
return bytes_transferred;
}
// Build a generic MIOP Header on the stack for this
// "number_of_packets_required". NOTE: We can save
// pointers and write to them later without byte swapping
// since in CORBA, the sender chooses the endianess.
char header_buffer[MIOP_DEFAULT_HEADER_SIZE + ACE_CDR::MAX_ALIGNMENT];
TAO_OutputCDR miop_hdr (header_buffer, sizeof header_buffer);
miop_hdr.write_octet_array (miop_magic, sizeof miop_magic);
// MIOP (Not GIOP) Version
miop_hdr.write_octet (((TAO_DEF_MIOP_MAJOR & 0xf) << 4) +
(TAO_DEF_MIOP_MINOR & 0xf));
// Write flags octet:
// Bit Description
// 0 Endian
// 1 Stop message flag (Assigned later)
// 2-7 Set to 0
CORBA::Octet *flags_field =
reinterpret_cast<CORBA::Octet *> (miop_hdr.current ()->wr_ptr ());
miop_hdr.write_octet (TAO_ENCAP_BYTE_ORDER);
CORBA::UShort *packet_length =
reinterpret_cast<CORBA::UShort *> (miop_hdr.current ()->wr_ptr ());
miop_hdr.write_short (0);
CORBA::ULong *packet_number =
reinterpret_cast<CORBA::ULong *> (miop_hdr.current ()->wr_ptr ());
miop_hdr.write_ulong (0u);
miop_hdr.write_ulong (number_of_packets_required);
if (!this->write_unique_id (miop_hdr))
{
ORBSVCS_DEBUG ((LM_ERROR,
ACE_TEXT ("TAO (%P|%t) - UIPMC_Transport[%d]::send, ")
ACE_TEXT ("error creating fragment MIOP header.\n"),
this->id ()));
return -1;
}
// Attempt to partition up the payload data sending each of the MIOP fragments
iovec this_fragment_iov[ACE_IOV_MAX];
UIPMC_Message_Block_Data_Iterator mb_iter (iov, iovcnt);
ACE_INET_Addr const &addr = this->connection_handler_->addr ();
for (*packet_number= 0u;
*packet_number < number_of_packets_required;
++*packet_number)
{
// The first iov for each packet points at the static MIOP header.
this_fragment_iov[0].iov_base = miop_hdr.current ()->rd_ptr ();
this_fragment_iov[0].iov_len = MIOP_DEFAULT_HEADER_SIZE;
int this_fragment_iovcnt = 1; // Just the MIOP Header so far!
u_long this_fragment_size= 0uL; // Just the payload data length (for now)
// Obtain the next fragment's payload data.
while (mb_iter.next_block (max_fragment_payload - this_fragment_size,
this_fragment_iov[this_fragment_iovcnt]))
{
// Increment the fragments length and iovcnt.
this_fragment_size +=
this_fragment_iov[this_fragment_iovcnt++].iov_len;
// Check if we have maxed out this fragment's payload.
if (this_fragment_size == max_fragment_payload)
break;
// Just a safety check for building iovec.
if (this_fragment_iovcnt >= ACE_IOV_MAX)
{
ORBSVCS_DEBUG ((LM_ERROR,
ACE_TEXT ("TAO (%P|%t) - UIPMC_Transport[%d]::send, ")
ACE_TEXT ("Too many iovec to create fragment.\n"),
this->id ()));
return -1;
}
} // While fragment is not complete
// Now we have the payload length for this fragment, update the MIOP header
*packet_length = static_cast<CORBA::UShort> (this_fragment_size);
if (*packet_number == number_of_packets_required-1uL)
*flags_field |= 0x02;
ssize_t already_sent = 0; // No data sent yet!
iovec *current_iov= this_fragment_iov;
for (this_fragment_size+= MIOP_DEFAULT_HEADER_SIZE; // Now includes MIOP header
this_fragment_size; // Still any data to send
this_fragment_size-= static_cast<u_long> (already_sent))
{
// Make sure we don't send our fragments too quickly
if (factory->enable_throttling ())
this->throttle_send_rate (
factory->max_fragment_rate (),
max_fragment_size,
this_fragment_size);
// Haven't sent some of the data yet, we need to adjust the fragments iov's
// to skip the data we have actually manage to send so far.
while (already_sent)
if (static_cast<u_long> (current_iov->iov_len) <= static_cast<u_long> (already_sent))
{
// This whole iov has been sent, simply skip over it
already_sent-= current_iov->iov_len;
--this_fragment_iovcnt;
++current_iov;
}
else
{
// This iov has been partially sent, adjust it's data
// to skip over those bytes already transmitted.
current_iov->iov_len -= static_cast<u_long> (already_sent);
current_iov->iov_base =
&static_cast<char *> (current_iov->iov_base)[already_sent];
break; // already_sent= 0;
}
// Ok now we attempt to actually send the fragment.
already_sent =
this->connection_handler_->peer ().send (
current_iov,
this_fragment_iovcnt,
addr);
if (already_sent < 0)
{
ORBSVCS_DEBUG ((LM_ERROR,
ACE_TEXT ("TAO (%P|%t) - UIPMC_Transport[%d]::send, ")
ACE_TEXT ("error sending data (Errno: '%m')\n"),
this->id ()));
return -1;
}
else if (TAO_debug_level &&
static_cast<u_long> (already_sent) != this_fragment_size)
{
ORBSVCS_DEBUG ((LM_DEBUG,
ACE_TEXT ("TAO (%P|%t) - UIPMC_Transport[%d]::send, ")
ACE_TEXT ("Partial fragment (%B/%u bytes), ")
ACE_TEXT ("reattempting remainder.\n"),
this->id (),
already_sent,
this_fragment_size));
}
// Keep a note of the number of bytes we have just buffered
if (factory->enable_throttling ())
this->total_bytes_outstanding_+= static_cast<u_long> (already_sent);
} // Keep sending the rest of the fragment
// Increment the number of bytes of payload transferred.
bytes_transferred += *packet_length;
if (9 <= TAO_debug_level)
{
char tmp[INET6_ADDRSTRLEN];
addr.get_host_addr (tmp, sizeof tmp);
ORBSVCS_DEBUG ((LM_DEBUG,
ACE_TEXT ("TAO (%P|%t) - UIPMC_Transport[%d]::send, ")
ACE_TEXT ("Sent %u bytes payload (fragment %u/%u) to <%C:%u>\n"),
this->id (),
*packet_length,
*packet_number + 1uL,
number_of_packets_required,
tmp,
addr.get_port_number ()));
}
} // Send next fragment
// Return total bytes transferred.
return bytes_transferred;
}
ssize_t
TAO_UIPMC_Transport::recv (
char *,
size_t,
const ACE_Time_Value *)
{
// Shouldn't ever be called on the client side.
ACE_ASSERT (0);
return -1;
}
int
TAO_UIPMC_Transport::register_handler (void)
{
// We never register the handler with the reactor
// as we never need to be informed about any incoming data,
// assuming we only use one-ways.
// If we would register and ICMP Messages would arrive, e.g
// due to a not reachable server, we would get informed - as this
// disturbs the general MIOP assumptions of not being
// interested in any network failures, we ignore ICMP messages.
return 0;
}
int
TAO_UIPMC_Transport::send_request (
TAO_Stub *stub,
TAO_ORB_Core *orb_core,
TAO_OutputCDR &stream,
TAO_Message_Semantics message_semantics,
ACE_Time_Value *max_wait_time)
{
if (this->ws_->sending_request (orb_core,
message_semantics) == -1 ||
this->send_message (stream,
stub,
0,
message_semantics,
max_wait_time) == -1)
{
return -1;
}
this->first_request_sent();
return 0;
}
int
TAO_UIPMC_Transport::send_message (
TAO_OutputCDR &stream,
TAO_Stub *stub,
TAO_ServerRequest *request,
TAO_Message_Semantics message_semantics,
ACE_Time_Value *max_wait_time)
{
// Format the message in the stream first
if (this->messaging_object ()->format_message (stream, stub, request) != 0)
{
return -1;
}
// Strictly speaking, should not need to loop here because the
// socket never gets set to a nonblocking mode ... some Linux
// versions seem to need it though. Leaving it costs little.
// This guarantees to send all data (bytes) or return an error.
ssize_t const n = this->send_message_shared (stub,
message_semantics,
stream.begin (),
max_wait_time);
if (n == -1)
{
if (TAO_debug_level)
ORBSVCS_DEBUG ((LM_ERROR,
ACE_TEXT ("TAO: (%P|%t) - UIPMC_Transport[%d]::")
ACE_TEXT ("send_message, closing transport %d after ")
ACE_TEXT ("fault (Errno: '%m')\n"),
this->id ()));
return -1;
}
return 1;
}
TAO_END_VERSIONED_NAMESPACE_DECL
|