Changeset 1104:757df62cadb6 in freeDiameter for libfdproto
- Timestamp:
- May 10, 2013, 7:49:19 PM (11 years ago)
- Branch:
- default
- Parents:
- 1103:d8591b1c56cd (diff), 1059:a1d6e1980132 (diff)
Note: this is a merge changeset, the changes displayed below correspond to the merge itself.
Use the (diff) links above to see all the changes relative to each parent. - Phase:
- public
- Files:
-
- 2 edited
Legend:
- Unmodified
- Added
- Removed
-
libfdproto/fifo.c
r1059 r1104 70 70 int highest;/* The highest count value for which h_cb has been called */ 71 71 int highest_ever; /* The max count value this queue has reached (for tweaking) */ 72 73 long long total_items; /* Cumulated number of items that went through this fifo (excluding current count), always increasing. */ 74 struct timespec total_time; /* Cumulated time all items spent in this queue, including blocking time (always growing, use deltas for monitoring) */ 75 struct timespec blocking_time; /* Cumulated time threads trying to post new items were blocked (queue full). */ 76 struct timespec last_time; /* For the last element retrieved from the queue, how long it take between posting (including blocking) and poping */ 77 78 }; 79 80 struct fifo_item { 81 struct fd_list item; 82 struct timespec posted_on; 72 83 }; 73 84 … … 108 119 109 120 /* Dump the content of a queue */ 110 void fd_fifo_dump(int level, char * name, struct fifo * queue, void (*dump_item)(int level, void * item)) 111 { 112 TRACE_ENTRY("%i %p %p %p", level, name, queue, dump_item); 113 114 if (!TRACE_BOOL(level)) 115 return; 116 117 fd_log_debug("Dumping queue '%s' (%p):", name ?: "?", queue); 121 DECLARE_FD_DUMP_PROTOTYPE(fd_fifo_dump, char * name, struct fifo * queue, fd_fifo_dump_item_cb dump_item) 122 { 123 FD_DUMP_HANDLE_OFFSET(); 124 125 if (name) { 126 CHECK_MALLOC_DO( fd_dump_extend( FD_DUMP_STD_PARAMS, "'%s'(@%p): ", name, queue), return NULL); 127 } else { 128 CHECK_MALLOC_DO( fd_dump_extend( FD_DUMP_STD_PARAMS, "{fifo}(@%p): ", queue), return NULL); 129 } 130 118 131 if (!CHECK_FIFO( queue )) { 119 fd_log_debug(" Queue invalid!"); 120 if (queue) 121 fd_log_debug(" (%x != %x)", queue->eyec, FIFO_EYEC); 122 return; 132 return fd_dump_extend(FD_DUMP_STD_PARAMS, "INVALID/NULL"); 123 133 } 124 134 125 135 CHECK_POSIX_DO( pthread_mutex_lock( &queue->mtx ), /* continue */ ); 126 fd_log_debug(" %d elements in queue / %d threads waiting", queue->count, queue->thrs);127 fd_log_debug(" %d elements max / %d threads waiting to push", queue->max, queue->thrs_push);128 fd_log_debug(" thresholds: %d / %d (h:%d), cb: %p,%p (%p), highest: %d",129 queue->high, queue->low, queue->highest,130 queue->h_cb, queue->l_cb, queue->data,131 queue->highest_ever);136 CHECK_MALLOC_DO( fd_dump_extend( FD_DUMP_STD_PARAMS, "items:%d,%d,%d threads:%d,%d stats:%lld/%ld.%06ld,%ld.%06ld,%ld.%06ld thresholds:%d,%d,%d,%p,%p,%p", 137 queue->count, queue->highest_ever, queue->max, 138 queue->thrs, queue->thrs_push, 139 queue->total_items,(long)queue->total_time.tv_sec,(long)(queue->total_time.tv_nsec/1000),(long)queue->blocking_time.tv_sec,(long)(queue->blocking_time.tv_nsec/1000),(long)queue->last_time.tv_sec,(long)(queue->last_time.tv_nsec/1000), 140 queue->high, queue->low, queue->highest, queue->h_cb, queue->l_cb, queue->data), 141 goto error); 132 142 133 143 if (dump_item) { … … 135 145 int i = 0; 136 146 for (li = queue->list.next; li != &queue->list; li = li->next) { 137 fd_log_debug(" [%i] item %p in fifo %p:", i++, li->o, queue); 138 (*dump_item)(level, li->o); 147 struct fifo_item * fi = (struct fifo_item *)li; 148 CHECK_MALLOC_DO( fd_dump_extend( FD_DUMP_STD_PARAMS, "\n [#%i](@%p)@%ld.%06ld: ", 149 i++, fi->item.o, (long)fi->posted_on.tv_sec,(long)(fi->posted_on.tv_nsec/1000)), 150 goto error); 151 CHECK_MALLOC_DO( (*dump_item)(FD_DUMP_STD_PARAMS, fi->item.o), goto error); 139 152 } 140 153 } 141 154 CHECK_POSIX_DO( pthread_mutex_unlock( &queue->mtx ), /* continue */ ); 142 155 156 return *buf; 157 error: 158 CHECK_POSIX_DO( pthread_mutex_unlock( &queue->mtx ), /* continue */ ); 159 return NULL; 143 160 } 144 161 … … 243 260 old->eyec = FIFO_EYEC; 244 261 262 /* Merge the stats in the new queue */ 263 new->total_items += old->total_items; 264 old->total_items = 0; 265 266 new->total_time.tv_nsec += old->total_time.tv_nsec; 267 new->total_time.tv_sec += old->total_time.tv_sec + (new->total_time.tv_nsec / 1000000000); 268 new->total_time.tv_nsec %= 1000000000; 269 old->total_time.tv_nsec = 0; 270 old->total_time.tv_sec = 0; 271 272 new->blocking_time.tv_nsec += old->blocking_time.tv_nsec; 273 new->blocking_time.tv_sec += old->blocking_time.tv_sec + (new->blocking_time.tv_nsec / 1000000000); 274 new->blocking_time.tv_nsec %= 1000000000; 275 old->blocking_time.tv_nsec = 0; 276 old->blocking_time.tv_sec = 0; 277 245 278 /* Unlock, we're done */ 246 279 CHECK_POSIX( pthread_mutex_unlock( &new->mtx ) ); … … 250 283 } 251 284 252 /* Get the length of the queue */ 253 int fd_fifo_length ( struct fifo * queue, int * length ) 254 { 255 TRACE_ENTRY( "%p %p", queue, length ); 285 /* Get the information on the queue */ 286 int fd_fifo_getstats( struct fifo * queue, int * current_count, int * limit_count, int * highest_count, long long * total_count, 287 struct timespec * total, struct timespec * blocking, struct timespec * last) 288 { 289 TRACE_ENTRY( "%p %p %p %p %p %p %p %p", queue, current_count, limit_count, highest_count, total_count, total, blocking, last); 256 290 257 291 /* Check the parameters */ 258 CHECK_PARAMS( CHECK_FIFO( queue ) && length);292 CHECK_PARAMS( CHECK_FIFO( queue ) ); 259 293 260 294 /* lock the queue */ 261 295 CHECK_POSIX( pthread_mutex_lock( &queue->mtx ) ); 262 296 263 /* Retrieve the count */ 264 *length = queue->count; 297 if (current_count) 298 *current_count = queue->count; 299 300 if (limit_count) 301 *limit_count = queue->max; 302 303 if (highest_count) 304 *highest_count = queue->highest_ever; 305 306 if (total_count) 307 *total_count = queue->total_items; 308 309 if (total) 310 memcpy(total, &queue->total_time, sizeof(struct timespec)); 311 312 if (blocking) 313 memcpy(blocking, &queue->blocking_time, sizeof(struct timespec)); 314 315 if (last) 316 memcpy(last, &queue->last_time, sizeof(struct timespec)); 265 317 266 318 /* Unlock */ … … 271 323 } 272 324 325 273 326 /* alternate version with no error checking */ 274 int fd_fifo_length _noerr( struct fifo * queue )327 int fd_fifo_length ( struct fifo * queue ) 275 328 { 276 329 if ( !CHECK_FIFO( queue ) ) … … 326 379 int fd_fifo_post_int ( struct fifo * queue, void ** item ) 327 380 { 328 struct f d_list* new;381 struct fifo_item * new; 329 382 int call_cb = 0; 383 struct timespec posted_on, queued_on; 330 384 331 385 TRACE_ENTRY( "%p %p", queue, item ); … … 333 387 /* Check the parameters */ 334 388 CHECK_PARAMS( CHECK_FIFO( queue ) && item && *item ); 389 390 /* Get the timing of this call */ 391 CHECK_SYS( clock_gettime(CLOCK_REALTIME, &posted_on) ); 335 392 336 393 /* lock the queue */ … … 353 410 354 411 /* Create a new list item */ 355 CHECK_MALLOC_DO( new = malloc (sizeof (struct f d_list)) , {412 CHECK_MALLOC_DO( new = malloc (sizeof (struct fifo_item)) , { 356 413 pthread_mutex_unlock( &queue->mtx ); 414 return ENOMEM; 357 415 } ); 358 416 359 fd_list_init( new, *item);417 fd_list_init(&new->item, *item); 360 418 *item = NULL; 361 419 362 420 /* Add the new item at the end */ 363 fd_list_insert_before( &queue->list, new);421 fd_list_insert_before( &queue->list, &new->item); 364 422 queue->count++; 365 423 if (queue->highest_ever < queue->count) … … 370 428 } 371 429 430 /* store timing */ 431 memcpy(&new->posted_on, &posted_on, sizeof(struct timespec)); 432 433 /* update queue timing info "blocking time" */ 434 { 435 long long blocked_ns; 436 CHECK_SYS( clock_gettime(CLOCK_REALTIME, &queued_on) ); 437 blocked_ns = (queued_on.tv_sec - posted_on.tv_sec) * 1000000000; 438 blocked_ns += (queued_on.tv_nsec - posted_on.tv_nsec); 439 blocked_ns += queue->blocking_time.tv_nsec; 440 queue->blocking_time.tv_sec += blocked_ns / 1000000000; 441 queue->blocking_time.tv_nsec = blocked_ns % 1000000000; 442 } 443 372 444 /* Signal if threads are asleep */ 373 445 if (queue->thrs > 0) { … … 394 466 { 395 467 void * ret = NULL; 396 struct fd_list * li; 468 struct fifo_item * fi; 469 struct timespec now; 397 470 398 471 ASSERT( ! FD_IS_LIST_EMPTY(&queue->list) ); 399 472 400 fd_list_unlink(li = queue->list.next); 473 fi = (struct fifo_item *)(queue->list.next); 474 ret = fi->item.o; 475 fd_list_unlink(&fi->item); 401 476 queue->count--; 402 ret = li->o; 403 free(li); 477 queue->total_items++; 478 479 /* Update the timings */ 480 CHECK_SYS_DO( clock_gettime(CLOCK_REALTIME, &now), goto skip_timing ); 481 { 482 long long elapsed = (now.tv_sec - fi->posted_on.tv_sec) * 1000000000; 483 elapsed += now.tv_nsec - fi->posted_on.tv_nsec; 484 485 queue->last_time.tv_sec = elapsed / 1000000000; 486 queue->last_time.tv_nsec = elapsed % 1000000000; 487 488 elapsed += queue->total_time.tv_nsec; 489 queue->total_time.tv_sec += elapsed / 1000000000; 490 queue->total_time.tv_nsec = elapsed % 1000000000; 491 } 492 skip_timing: 493 free(fi); 404 494 405 495 if (queue->thrs_push) { … … 509 599 return EPIPE; 510 600 } 511 601 512 602 if (queue->count > 0) { 513 603 /* There are items in the queue, so pick the first one */ -
libfdproto/fifo.c
r1103 r1104 579 579 static int fifo_tget ( struct fifo * queue, void ** item, int istimed, const struct timespec *abstime) 580 580 { 581 int timedout = 0;582 581 int call_cb = 0; 582 int ret = 0; 583 583 584 584 /* Check the parameters */ … … 605 605 call_cb = test_l_cb(queue); 606 606 } else { 607 int ret = 0;608 607 /* We have to wait for a new item */ 609 608 queue->thrs++ ; … … 619 618 goto awaken; /* test for spurious wake-ups */ 620 619 621 if (istimed && (ret == ETIMEDOUT)) { 622 timedout = 1; 623 } else { 624 /* Unexpected error condition (means we need to debug) */ 625 ASSERT( ret == 0 /* never true */ ); 626 } 620 /* otherwise (ETIMEDOUT / other error) just continue */ 627 621 } 628 622 … … 635 629 636 630 /* Done */ 637 return timedout ? ETIMEDOUT : 0;631 return ret; 638 632 } 639 633
Note: See TracChangeset
for help on using the changeset viewer.