| … | … | @@ -30,7 +30,6 @@ |
| 30 | 30 | #include <mutex> |
| 31 | 31 | #include <thread> |
| 32 | 32 | #include <vector> |
| 33 | #include <unistd.h> /* for usleep() */ | |
| 34 | 33 | |
| 35 | 34 | #include <opencog/util/concurrent_set.h> |
| 36 | 35 | #include <opencog/util/exceptions.h> |
| … | … | @@ -120,11 +119,6 @@ namespace opencog |
| 120 | 119 | * barrier() call needs to do is to be a fence, ensuring that everything |
| 121 | 120 | * before really is before everything after. It didn't need to actually |
| 122 | 121 | * drain everything. |
| 123 | * | |
| 124 | * Another issue is that the assorted loops to drain queues use | |
| 125 | * spin-loops and usleeps. I think this is mostly harmless, but is | |
| 126 | * impure, and should be replaced by CV's. Doing so becomes much easier | |
| 127 | * once C++20 is widely available, as it has CV's on std::atomic ints. | |
| 128 | 122 | */ |
| 129 | 123 | template<typename Writer, typename Element> |
| 130 | 124 | class async_buffer |
| … | … | @@ -339,11 +333,11 @@ void async_buffer<Writer, Element>::stop_writer_threads() |
| 339 | 333 | |
| 340 | 334 | _stopping_writers = true; |
| 341 | 335 | |
| 342 | // Spin a while, until the writer threads are (mostly) done. | |
| 336 | // Wait until the writer threads are (mostly) done. | |
| 343 | 337 | while (0 < _pending) |
| 344 | 338 | { |
| 345 | // std::this_thread::sleep_for(std::chrono::milliseconds(1)); | |
| 346 | usleep(1000); | |
| 339 | unsigned long pend = _pending.load(); | |
| 340 | if (pend > 0) _pending.wait(pend); | |
| 347 | 341 | } |
| 348 | 342 | |
| 349 | 343 | // Now tell all the threads that they are done. |
| … | … | @@ -385,10 +379,12 @@ void async_buffer<Writer, Element>::drain() |
| 385 | 379 | _stall_writers = false; |
| 386 | 380 | _flush_count++; |
| 387 | 381 | |
| 388 | // XXX TODO - when C++20 becomes widely available, | |
| 389 | // replace this loop by _pending.wait(0) | |
| 382 | // Wait for all pending work to complete | |
| 390 | 383 | while (0 < _pending) |
| 391 | std::this_thread::sleep_for(std::chrono::milliseconds(1)); | |
| 384 | { | |
| 385 | unsigned long pend = _pending.load(); | |
| 386 | if (pend > 0) _pending.wait(pend); | |
| 387 | } | |
| 392 | 388 | |
| 393 | 389 | _stall_writers = save_stall; |
| 394 | 390 | } |
| … | … | @@ -468,7 +464,10 @@ void async_buffer<Writer, Element>::barrier(const Element& elt) |
| 468 | 464 | |
| 469 | 465 | // Wait for all threads to complete their barrier work |
| 470 | 466 | while (_barrier_remaining.load() > 0) |
| 471 | std::this_thread::sleep_for(std::chrono::milliseconds(1)); | |
| 467 | { | |
| 468 | unsigned rem = _barrier_remaining.load(); | |
| 469 | if (rem > 0) _barrier_remaining.wait(rem); | |
| 470 | } | |
| 472 | 471 | } |
| 473 | 472 | |
| 474 | 473 | /// A single write thread. Reads elements from set, and invokes the |
| … | … | @@ -490,7 +489,9 @@ void async_buffer<Writer, Element>::write_loop() |
| 490 | 489 | while (_barrier_remaining.load() > 0 and |
| 491 | 490 | my_last_phase >= _barrier_phase.load()) |
| 492 | 491 | { |
| 493 | std::this_thread::sleep_for(std::chrono::milliseconds(1)); | |
| 492 | unsigned rem = _barrier_remaining.load(); | |
| 493 | if (rem > 0 and my_last_phase >= _barrier_phase.load()) | |
| 494 | _barrier_remaining.wait(rem); | |
| 494 | 495 | } |
| 495 | 496 | |
| 496 | 497 | // Do nothing, if asked to stall. |
| … | … | @@ -509,6 +510,7 @@ void async_buffer<Writer, Element>::write_loop() |
| 509 | 510 | my_last_phase = current_phase; |
| 510 | 511 | (_writer->*_do_write)(elt); |
| 511 | 512 | _barrier_remaining.fetch_sub(1); |
| 513 | _barrier_remaining.notify_all(); | |
| 512 | 514 | } |
| 513 | 515 | else |
| 514 | 516 | { |
| … | … | @@ -518,6 +520,7 @@ void async_buffer<Writer, Element>::write_loop() |
| 518 | 520 | |
| 519 | 521 | _busy_writers --; |
| 520 | 522 | _pending --; |
| 523 | _pending.notify_all(); | |
| 521 | 524 | } |
| 522 | 525 | } |
| 523 | 526 | catch (typename concurrent_set<Element>::Canceled& e) |
| … | … | @@ -94,11 +94,6 @@ namespace opencog |
| 94 | 94 | * barrier() call needs to do is to be a fence, ensuring that everything |
| 95 | 95 | * before really is before everything after. It didn't need to actually |
| 96 | 96 | * drain everything. |
| 97 | * | |
| 98 | * Another issue is that the assorted loops to drain queues use | |
| 99 | * spin-loops and usleeps. I think this is mostly harmless, but is | |
| 100 | * impure, and should be replaced by CV's. Doing so becomes much easier | |
| 101 | * once C++20 is widely available, as it has CV's on std::atomic ints. | |
| 102 | 97 | */ |
| 103 | 98 | template<typename Writer, typename Element> |
| 104 | 99 | class async_caller |
| … | … | @@ -258,11 +253,11 @@ void async_caller<Writer, Element>::stop_writer_threads() |
| 258 | 253 | |
| 259 | 254 | _stopping_writers = true; |
| 260 | 255 | |
| 261 | // Spin a while, until the writer threads are (mostly) done. | |
| 256 | // Wait until the writer threads are (mostly) done. | |
| 262 | 257 | while (0 < _pending) |
| 263 | 258 | { |
| 264 | std::this_thread::sleep_for(std::chrono::milliseconds(1)); | |
| 265 | // usleep(1000); | |
| 259 | unsigned long pend = _pending.load(); | |
| 260 | if (pend > 0) _pending.wait(pend); | |
| 266 | 261 | } |
| 267 | 262 | |
| 268 | 263 | // Now tell all the threads that they are done. |
| … | … | @@ -302,10 +297,12 @@ void async_caller<Writer, Element>::drain() |
| 302 | 297 | { |
| 303 | 298 | _flush_count++; |
| 304 | 299 | |
| 305 | // XXX TODO - when C++20 becomes widely available, | |
| 306 | // replace this loop by _pending.wait(0) | |
| 300 | // Wait for all pending work to complete | |
| 307 | 301 | while (0 < _pending) |
| 308 | std::this_thread::sleep_for(std::chrono::milliseconds(1)); | |
| 302 | { | |
| 303 | unsigned long pend = _pending.load(); | |
| 304 | if (pend > 0) _pending.wait(pend); | |
| 305 | } | |
| 309 | 306 | } |
| 310 | 307 | |
| 311 | 308 | |
| … | … | @@ -385,7 +382,10 @@ void async_caller<Writer, Element>::barrier(const Element& elt) |
| 385 | 382 | |
| 386 | 383 | // Wait for all threads to complete their barrier work |
| 387 | 384 | while (_barrier_remaining.load() > 0) |
| 388 | std::this_thread::sleep_for(std::chrono::milliseconds(1)); | |
| 385 | { | |
| 386 | unsigned rem = _barrier_remaining.load(); | |
| 387 | if (rem > 0) _barrier_remaining.wait(rem); | |
| 388 | } | |
| 389 | 389 | } |
| 390 | 390 | |
| 391 | 391 | /// A single write thread. Reads elements from queue, and invokes the |
| … | … | @@ -407,7 +407,9 @@ void async_caller<Writer, Element>::write_loop() |
| 407 | 407 | while (_barrier_remaining.load() > 0 and |
| 408 | 408 | my_last_phase >= _barrier_phase.load()) |
| 409 | 409 | { |
| 410 | std::this_thread::sleep_for(std::chrono::milliseconds(1)); | |
| 410 | unsigned rem = _barrier_remaining.load(); | |
| 411 | if (rem > 0 and my_last_phase >= _barrier_phase.load()) | |
| 412 | _barrier_remaining.wait(rem); | |
| 411 | 413 | } |
| 412 | 414 | |
| 413 | 415 | Element elt = _store_queue.value_pop(); |
| … | … | @@ -420,6 +422,7 @@ void async_caller<Writer, Element>::write_loop() |
| 420 | 422 | my_last_phase = current_phase; |
| 421 | 423 | (_writer->*_do_write)(elt); |
| 422 | 424 | _barrier_remaining.fetch_sub(1); |
| 425 | _barrier_remaining.notify_all(); | |
| 423 | 426 | } |
| 424 | 427 | else |
| 425 | 428 | { |
| … | … | @@ -429,6 +432,7 @@ void async_caller<Writer, Element>::write_loop() |
| 429 | 432 | |
| 430 | 433 | _busy_writers --; |
| 431 | 434 | _pending --; |
| 435 | _pending.notify_all(); | |
| 432 | 436 | } |
| 433 | 437 | } |
| 434 | 438 | catch (typename concurrent_queue<Element>::Canceled& e) |