Explore Enterprise Education Gitee Premium Gitee AI AI teammates
Fetch the repository succeeded.
Create your Gitee Account
Explore and code with more than 14 million developers,Free private repositories !:)
Sign up
Already have an account? Sign in
文件
master
Branches (1)
master
This repository doesn't specify license. Please pay attention to the specific project description and its upstream code dependency when using it.
The license selected for the repository is subject to the license used by the main branch of the repository.
master
Branches (1)
master
Clone or Download
Clone/Download
Prompt
To download the code, please copy the following command and execute it in the terminal
To ensure that your submitted code identity is correctly recognized by Gitee, please execute the following command.
When using the SSH protocol for the first time to clone or push code, follow the prompts below to complete the SSH configuration.
1 Generate RSA keys.
2 Obtain the content of the RSA public key and configure it in SSH Public Keys
To use SVN on Gitee, please visit the usage guide
When using the HTTPS protocol, the command line will prompt for account and password verification as follows. For security reasons, Gitee recommends configure and use personal access tokens instead of login passwords for cloning, pushing, and other operations.
Username for 'https://gitee.com': userName
Password for 'https://userName@gitee.com': # Private Token
master
Branches (1)
master
jdk11.0.2lib.src.java.base.java
/
util
/
concurrent
/
LinkedTransferQueue.java
jdk11.0.2lib.src.java.base.java
/
util
/
concurrent
/
LinkedTransferQueue.java
LinkedTransferQueue.java 66.95 KB
Copy Edit Raw Blame History
gwdcode authored 2020年10月30日 22:13 +08:00 . 常用普通java包(jdk11.0.2)
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 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523 524 525 526 527 528 529 530 531 532 533 534 535 536 537 538 539 540 541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556 557 558 559 560 561 562 563 564 565 566 567 568 569 570 571 572 573 574 575 576 577 578 579 580 581 582 583 584 585 586 587 588 589 590 591 592 593 594 595 596 597 598 599 600 601 602 603 604 605 606 607 608 609 610 611 612 613 614 615 616 617 618 619 620 621 622 623 624 625 626 627 628 629 630 631 632 633 634 635 636 637 638 639 640 641 642 643 644 645 646 647 648 649 650 651 652 653 654 655 656 657 658 659 660 661 662 663 664 665 666 667 668 669 670 671 672 673 674 675 676 677 678 679 680 681 682 683 684 685 686 687 688 689 690 691 692 693 694 695 696 697 698 699 700 701 702 703 704 705 706 707 708 709 710 711 712 713 714 715 716 717 718 719 720 721 722 723 724 725 726 727 728 729 730 731 732 733 734 735 736 737 738 739 740 741 742 743 744 745 746 747 748 749 750 751 752 753 754 755 756 757 758 759 760 761 762 763 764 765 766 767 768 769 770 771 772 773 774 775 776 777 778 779 780 781 782 783 784 785 786 787 788 789 790 791 792 793 794 795 796 797 798 799 800 801 802 803 804 805 806 807 808 809 810 811 812 813 814 815 816 817 818 819 820 821 822 823 824 825 826 827 828 829 830 831 832 833 834 835 836 837 838 839 840 841 842 843 844 845 846 847 848 849 850 851 852 853 854 855 856 857 858 859 860 861 862 863 864 865 866 867 868 869 870 871 872 873 874 875 876 877 878 879 880 881 882 883 884 885 886 887 888 889 890 891 892 893 894 895 896 897 898 899 900 901 902 903 904 905 906 907 908 909 910 911 912 913 914 915 916 917 918 919 920 921 922 923 924 925 926 927 928 929 930 931 932 933 934 935 936 937 938 939 940 941 942 943 944 945 946 947 948 949 950 951 952 953 954 955 956 957 958 959 960 961 962 963 964 965 966 967 968 969 970 971 972 973 974 975 976 977 978 979 980 981 982 983 984 985 986 987 988 989 990 991 992 993 994 995 996 997 998 999 1000 1001 1002 1003 1004 1005 1006 1007 1008 1009 1010 1011 1012 1013 1014 1015 1016 1017 1018 1019 1020 1021 1022 1023 1024 1025 1026 1027 1028 1029 1030 1031 1032 1033 1034 1035 1036 1037 1038 1039 1040 1041 1042 1043 1044 1045 1046 1047 1048 1049 1050 1051 1052 1053 1054 1055 1056 1057 1058 1059 1060 1061 1062 1063 1064 1065 1066 1067 1068 1069 1070 1071 1072 1073 1074 1075 1076 1077 1078 1079 1080 1081 1082 1083 1084 1085 1086 1087 1088 1089 1090 1091 1092 1093 1094 1095 1096 1097 1098 1099 1100 1101 1102 1103 1104 1105 1106 1107 1108 1109 1110 1111 1112 1113 1114 1115 1116 1117 1118 1119 1120 1121 1122 1123 1124 1125 1126 1127 1128 1129 1130 1131 1132 1133 1134 1135 1136 1137 1138 1139 1140 1141 1142 1143 1144 1145 1146 1147 1148 1149 1150 1151 1152 1153 1154 1155 1156 1157 1158 1159 1160 1161 1162 1163 1164 1165 1166 1167 1168 1169 1170 1171 1172 1173 1174 1175 1176 1177 1178 1179 1180 1181 1182 1183 1184 1185 1186 1187 1188 1189 1190 1191 1192 1193 1194 1195 1196 1197 1198 1199 1200 1201 1202 1203 1204 1205 1206 1207 1208 1209 1210 1211 1212 1213 1214 1215 1216 1217 1218 1219 1220 1221 1222 1223 1224 1225 1226 1227 1228 1229 1230 1231 1232 1233 1234 1235 1236 1237 1238 1239 1240 1241 1242 1243 1244 1245 1246 1247 1248 1249 1250 1251 1252 1253 1254 1255 1256 1257 1258 1259 1260 1261 1262 1263 1264 1265 1266 1267 1268 1269 1270 1271 1272 1273 1274 1275 1276 1277 1278 1279 1280 1281 1282 1283 1284 1285 1286 1287 1288 1289 1290 1291 1292 1293 1294 1295 1296 1297 1298 1299 1300 1301 1302 1303 1304 1305 1306 1307 1308 1309 1310 1311 1312 1313 1314 1315 1316 1317 1318 1319 1320 1321 1322 1323 1324 1325 1326 1327 1328 1329 1330 1331 1332 1333 1334 1335 1336 1337 1338 1339 1340 1341 1342 1343 1344 1345 1346 1347 1348 1349 1350 1351 1352 1353 1354 1355 1356 1357 1358 1359 1360 1361 1362 1363 1364 1365 1366 1367 1368 1369 1370 1371 1372 1373 1374 1375 1376 1377 1378 1379 1380 1381 1382 1383 1384 1385 1386 1387 1388 1389 1390 1391 1392 1393 1394 1395 1396 1397 1398 1399 1400 1401 1402 1403 1404 1405 1406 1407 1408 1409 1410 1411 1412 1413 1414 1415 1416 1417 1418 1419 1420 1421 1422 1423 1424 1425 1426 1427 1428 1429 1430 1431 1432 1433 1434 1435 1436 1437 1438 1439 1440 1441 1442 1443 1444 1445 1446 1447 1448 1449 1450 1451 1452 1453 1454 1455 1456 1457 1458 1459 1460 1461 1462 1463 1464 1465 1466 1467 1468 1469 1470 1471 1472 1473 1474 1475 1476 1477 1478 1479 1480 1481 1482 1483 1484 1485 1486 1487 1488 1489 1490 1491 1492 1493 1494 1495 1496 1497 1498 1499 1500 1501 1502 1503 1504 1505 1506 1507 1508 1509 1510 1511 1512 1513 1514 1515 1516 1517 1518 1519 1520 1521 1522 1523 1524 1525 1526 1527 1528 1529 1530 1531 1532 1533 1534 1535 1536 1537 1538 1539 1540 1541 1542 1543 1544 1545 1546 1547 1548 1549 1550 1551 1552 1553 1554 1555 1556 1557 1558 1559 1560 1561 1562 1563 1564 1565 1566 1567 1568 1569 1570 1571 1572 1573 1574 1575 1576 1577 1578 1579 1580 1581 1582 1583 1584 1585 1586 1587 1588 1589 1590 1591 1592 1593 1594 1595 1596 1597 1598 1599 1600 1601 1602 1603 1604 1605 1606 1607 1608 1609 1610 1611 1612 1613 1614 1615 1616 1617 1618 1619 1620 1621 1622 1623 1624 1625 1626 1627 1628 1629 1630 1631 1632 1633 1634 1635 1636 1637 1638 1639 1640 1641 1642 1643 1644 1645 1646 1647 1648 1649 1650 1651 1652 1653 1654 1655 1656 1657 1658 1659 1660 1661 1662 1663 1664 1665 1666 1667 1668 1669 1670 1671 1672 1673 1674 1675 1676 1677 1678 1679 1680 1681 1682 1683 1684 1685 1686 1687 1688 1689 1690 1691 1692 1693 1694 1695 1696 1697 1698 1699 1700 1701 1702 1703 1704 1705 1706 1707 1708 1709 1710 1711 1712 1713 1714 1715 1716 1717 1718 1719 1720 1721 1722 1723 1724 1725 1726 1727 1728 1729 1730 1731 1732 1733 1734 1735 1736 1737 1738 1739 1740 1741 1742 1743 1744 1745 1746 1747 1748 1749
/*
* ORACLE PROPRIETARY/CONFIDENTIAL. Use is subject to license terms.
*
*
*
*
*
*
*
*
*
*
*
*
*
*
*
*
*
*
*
*
*/
/*
*
*
*
*
*
* Written by Doug Lea with assistance from members of JCP JSR-166
* Expert Group and released to the public domain, as explained at
* http://creativecommons.org/publicdomain/zero/1.0/
*/
package java.util.concurrent;
import java.lang.invoke.MethodHandles;
import java.lang.invoke.VarHandle;
import java.util.AbstractQueue;
import java.util.Arrays;
import java.util.Collection;
import java.util.Iterator;
import java.util.NoSuchElementException;
import java.util.Objects;
import java.util.Queue;
import java.util.Spliterator;
import java.util.Spliterators;
import java.util.concurrent.locks.LockSupport;
import java.util.function.Consumer;
import java.util.function.Predicate;
/**
* An unbounded {@link TransferQueue} based on linked nodes.
* This queue orders elements FIFO (first-in-first-out) with respect
* to any given producer. The <em>head</em> of the queue is that
* element that has been on the queue the longest time for some
* producer. The <em>tail</em> of the queue is that element that has
* been on the queue the shortest time for some producer.
*
* <p>Beware that, unlike in most collections, the {@code size} method
* is <em>NOT</em> a constant-time operation. Because of the
* asynchronous nature of these queues, determining the current number
* of elements requires a traversal of the elements, and so may report
* inaccurate results if this collection is modified during traversal.
*
* <p>Bulk operations that add, remove, or examine multiple elements,
* such as {@link #addAll}, {@link #removeIf} or {@link #forEach},
* are <em>not</em> guaranteed to be performed atomically.
* For example, a {@code forEach} traversal concurrent with an {@code
* addAll} operation might observe only some of the added elements.
*
* <p>This class and its iterator implement all of the <em>optional</em>
* methods of the {@link Collection} and {@link Iterator} interfaces.
*
* <p>Memory consistency effects: As with other concurrent
* collections, actions in a thread prior to placing an object into a
* {@code LinkedTransferQueue}
* <a href="package-summary.html#MemoryVisibility"><i>happen-before</i></a>
* actions subsequent to the access or removal of that element from
* the {@code LinkedTransferQueue} in another thread.
*
* <p>This class is a member of the
* <a href="{@docRoot}/java.base/java/util/package-summary.html#CollectionsFramework">
* Java Collections Framework</a>.
*
* @since 1.7
* @author Doug Lea
* @param <E> the type of elements held in this queue
*/
public class LinkedTransferQueue<E> extends AbstractQueue<E>
implements TransferQueue<E>, java.io.Serializable {
private static final long serialVersionUID = -3223113410248163686L;
/*
* *** Overview of Dual Queues with Slack ***
*
* Dual Queues, introduced by Scherer and Scott
* (http://www.cs.rochester.edu/~scott/papers/2004_DISC_dual_DS.pdf)
* are (linked) queues in which nodes may represent either data or
* requests. When a thread tries to enqueue a data node, but
* encounters a request node, it instead "matches" and removes it;
* and vice versa for enqueuing requests. Blocking Dual Queues
* arrange that threads enqueuing unmatched requests block until
* other threads provide the match. Dual Synchronous Queues (see
* Scherer, Lea, & Scott
* http://www.cs.rochester.edu/u/scott/papers/2009_Scherer_CACM_SSQ.pdf)
* additionally arrange that threads enqueuing unmatched data also
* block. Dual Transfer Queues support all of these modes, as
* dictated by callers.
*
* A FIFO dual queue may be implemented using a variation of the
* Michael & Scott (M&S) lock-free queue algorithm
* (http://www.cs.rochester.edu/~scott/papers/1996_PODC_queues.pdf).
* It maintains two pointer fields, "head", pointing to a
* (matched) node that in turn points to the first actual
* (unmatched) queue node (or null if empty); and "tail" that
* points to the last node on the queue (or again null if
* empty). For example, here is a possible queue with four data
* elements:
*
* head tail
* | |
* v v
* M -> U -> U -> U -> U
*
* The M&S queue algorithm is known to be prone to scalability and
* overhead limitations when maintaining (via CAS) these head and
* tail pointers. This has led to the development of
* contention-reducing variants such as elimination arrays (see
* Moir et al http://portal.acm.org/citation.cfm?id=1074013) and
* optimistic back pointers (see Ladan-Mozes & Shavit
* http://people.csail.mit.edu/edya/publications/OptimisticFIFOQueue-journal.pdf).
* However, the nature of dual queues enables a simpler tactic for
* improving M&S-style implementations when dual-ness is needed.
*
* In a dual queue, each node must atomically maintain its match
* status. While there are other possible variants, we implement
* this here as: for a data-mode node, matching entails CASing an
* "item" field from a non-null data value to null upon match, and
* vice-versa for request nodes, CASing from null to a data
* value. (Note that the linearization properties of this style of
* queue are easy to verify -- elements are made available by
* linking, and unavailable by matching.) Compared to plain M&S
* queues, this property of dual queues requires one additional
* successful atomic operation per enq/deq pair. But it also
* enables lower cost variants of queue maintenance mechanics. (A
* variation of this idea applies even for non-dual queues that
* support deletion of interior elements, such as
* j.u.c.ConcurrentLinkedQueue.)
*
* Once a node is matched, its match status can never again
* change. We may thus arrange that the linked list of them
* contain a prefix of zero or more matched nodes, followed by a
* suffix of zero or more unmatched nodes. (Note that we allow
* both the prefix and suffix to be zero length, which in turn
* means that we do not use a dummy header.) If we were not
* concerned with either time or space efficiency, we could
* correctly perform enqueue and dequeue operations by traversing
* from a pointer to the initial node; CASing the item of the
* first unmatched node on match and CASing the next field of the
* trailing node on appends. While this would be a terrible idea
* in itself, it does have the benefit of not requiring ANY atomic
* updates on head/tail fields.
*
* We introduce here an approach that lies between the extremes of
* never versus always updating queue (head and tail) pointers.
* This offers a tradeoff between sometimes requiring extra
* traversal steps to locate the first and/or last unmatched
* nodes, versus the reduced overhead and contention of fewer
* updates to queue pointers. For example, a possible snapshot of
* a queue is:
*
* head tail
* | |
* v v
* M -> M -> U -> U -> U -> U
*
* The best value for this "slack" (the targeted maximum distance
* between the value of "head" and the first unmatched node, and
* similarly for "tail") is an empirical matter. We have found
* that using very small constants in the range of 1-3 work best
* over a range of platforms. Larger values introduce increasing
* costs of cache misses and risks of long traversal chains, while
* smaller values increase CAS contention and overhead.
*
* Dual queues with slack differ from plain M&S dual queues by
* virtue of only sometimes updating head or tail pointers when
* matching, appending, or even traversing nodes; in order to
* maintain a targeted slack. The idea of "sometimes" may be
* operationalized in several ways. The simplest is to use a
* per-operation counter incremented on each traversal step, and
* to try (via CAS) to update the associated queue pointer
* whenever the count exceeds a threshold. Another, that requires
* more overhead, is to use random number generators to update
* with a given probability per traversal step.
*
* In any strategy along these lines, because CASes updating
* fields may fail, the actual slack may exceed targeted slack.
* However, they may be retried at any time to maintain targets.
* Even when using very small slack values, this approach works
* well for dual queues because it allows all operations up to the
* point of matching or appending an item (hence potentially
* allowing progress by another thread) to be read-only, thus not
* introducing any further contention. As described below, we
* implement this by performing slack maintenance retries only
* after these points.
*
* As an accompaniment to such techniques, traversal overhead can
* be further reduced without increasing contention of head
* pointer updates: Threads may sometimes shortcut the "next" link
* path from the current "head" node to be closer to the currently
* known first unmatched node, and similarly for tail. Again, this
* may be triggered with using thresholds or randomization.
*
* These ideas must be further extended to avoid unbounded amounts
* of costly-to-reclaim garbage caused by the sequential "next"
* links of nodes starting at old forgotten head nodes: As first
* described in detail by Boehm
* (http://portal.acm.org/citation.cfm?doid=503272.503282), if a GC
* delays noticing that any arbitrarily old node has become
* garbage, all newer dead nodes will also be unreclaimed.
* (Similar issues arise in non-GC environments.) To cope with
* this in our implementation, upon CASing to advance the head
* pointer, we set the "next" link of the previous head to point
* only to itself; thus limiting the length of chains of dead nodes.
* (We also take similar care to wipe out possibly garbage
* retaining values held in other Node fields.) However, doing so
* adds some further complexity to traversal: If any "next"
* pointer links to itself, it indicates that the current thread
* has lagged behind a head-update, and so the traversal must
* continue from the "head". Traversals trying to find the
* current tail starting from "tail" may also encounter
* self-links, in which case they also continue at "head".
*
* It is tempting in slack-based scheme to not even use CAS for
* updates (similarly to Ladan-Mozes & Shavit). However, this
* cannot be done for head updates under the above link-forgetting
* mechanics because an update may leave head at a detached node.
* And while direct writes are possible for tail updates, they
* increase the risk of long retraversals, and hence long garbage
* chains, which can be much more costly than is worthwhile
* considering that the cost difference of performing a CAS vs
* write is smaller when they are not triggered on each operation
* (especially considering that writes and CASes equally require
* additional GC bookkeeping ("write barriers") that are sometimes
* more costly than the writes themselves because of contention).
*
* *** Overview of implementation ***
*
* We use a threshold-based approach to updates, with a slack
* threshold of two -- that is, we update head/tail when the
* current pointer appears to be two or more steps away from the
* first/last node. The slack value is hard-wired: a path greater
* than one is naturally implemented by checking equality of
* traversal pointers except when the list has only one element,
* in which case we keep slack threshold at one. Avoiding tracking
* explicit counts across method calls slightly simplifies an
* already-messy implementation. Using randomization would
* probably work better if there were a low-quality dirt-cheap
* per-thread one available, but even ThreadLocalRandom is too
* heavy for these purposes.
*
* With such a small slack threshold value, it is not worthwhile
* to augment this with path short-circuiting (i.e., unsplicing
* interior nodes) except in the case of cancellation/removal (see
* below).
*
* All enqueue/dequeue operations are handled by the single method
* "xfer" with parameters indicating whether to act as some form
* of offer, put, poll, take, or transfer (each possibly with
* timeout). The relative complexity of using one monolithic
* method outweighs the code bulk and maintenance problems of
* using separate methods for each case.
*
* Operation consists of up to two phases. The first is implemented
* in method xfer, the second in method awaitMatch.
*
* 1. Traverse until matching or appending (method xfer)
*
* Conceptually, we simply traverse all nodes starting from head.
* If we encounter an unmatched node of opposite mode, we match
* it and return, also updating head (by at least 2 hops) to
* one past the matched node (or the node itself if it's the
* pinned trailing node). Traversals also check for the
* possibility of falling off-list, in which case they restart.
*
* If the trailing node of the list is reached, a match is not
* possible. If this call was untimed poll or tryTransfer
* (argument "how" is NOW), return empty-handed immediately.
* Else a new node is CAS-appended. On successful append, if
* this call was ASYNC (e.g. offer), an element was
* successfully added to the end of the queue and we return.
*
* Of course, this naive traversal is O(n) when no match is
* possible. We optimize the traversal by maintaining a tail
* pointer, which is expected to be "near" the end of the list.
* It is only safe to fast-forward to tail (in the presence of
* arbitrary concurrent changes) if it is pointing to a node of
* the same mode, even if it is dead (in this case no preceding
* node could still be matchable by this traversal). If we
* need to restart due to falling off-list, we can again
* fast-forward to tail, but only if it has changed since the
* last traversal (else we might loop forever). If tail cannot
* be used, traversal starts at head (but in this case we
* expect to be able to match near head). As with head, we
* CAS-advance the tail pointer by at least two hops.
*
* 2. Await match or cancellation (method awaitMatch)
*
* Wait for another thread to match node; instead cancelling if
* the current thread was interrupted or the wait timed out. On
* multiprocessors, we use front-of-queue spinning: If a node
* appears to be the first unmatched node in the queue, it
* spins a bit before blocking. In either case, before blocking
* it tries to unsplice any nodes between the current "head"
* and the first unmatched node.
*
* Front-of-queue spinning vastly improves performance of
* heavily contended queues. And so long as it is relatively
* brief and "quiet", spinning does not much impact performance
* of less-contended queues. During spins threads check their
* interrupt status and generate a thread-local random number
* to decide to occasionally perform a Thread.yield. While
* yield has underdefined specs, we assume that it might help,
* and will not hurt, in limiting impact of spinning on busy
* systems. We also use smaller (1/2) spins for nodes that are
* not known to be front but whose predecessors have not
* blocked -- these "chained" spins avoid artifacts of
* front-of-queue rules which otherwise lead to alternating
* nodes spinning vs blocking. Further, front threads that
* represent phase changes (from data to request node or vice
* versa) compared to their predecessors receive additional
* chained spins, reflecting longer paths typically required to
* unblock threads during phase changes.
*
*
* ** Unlinking removed interior nodes **
*
* In addition to minimizing garbage retention via self-linking
* described above, we also unlink removed interior nodes. These
* may arise due to timed out or interrupted waits, or calls to
* remove(x) or Iterator.remove. Normally, given a node that was
* at one time known to be the predecessor of some node s that is
* to be removed, we can unsplice s by CASing the next field of
* its predecessor if it still points to s (otherwise s must
* already have been removed or is now offlist). But there are two
* situations in which we cannot guarantee to make node s
* unreachable in this way: (1) If s is the trailing node of list
* (i.e., with null next), then it is pinned as the target node
* for appends, so can only be removed later after other nodes are
* appended. (2) We cannot necessarily unlink s given a
* predecessor node that is matched (including the case of being
* cancelled): the predecessor may already be unspliced, in which
* case some previous reachable node may still point to s.
* (For further explanation see Herlihy & Shavit "The Art of
* Multiprocessor Programming" chapter 9). Although, in both
* cases, we can rule out the need for further action if either s
* or its predecessor are (or can be made to be) at, or fall off
* from, the head of list.
*
* Without taking these into account, it would be possible for an
* unbounded number of supposedly removed nodes to remain reachable.
* Situations leading to such buildup are uncommon but can occur
* in practice; for example when a series of short timed calls to
* poll repeatedly time out at the trailing node but otherwise
* never fall off the list because of an untimed call to take() at
* the front of the queue.
*
* When these cases arise, rather than always retraversing the
* entire list to find an actual predecessor to unlink (which
* won't help for case (1) anyway), we record a conservative
* estimate of possible unsplice failures (in "sweepVotes").
* We trigger a full sweep when the estimate exceeds a threshold
* ("SWEEP_THRESHOLD") indicating the maximum number of estimated
* removal failures to tolerate before sweeping through, unlinking
* cancelled nodes that were not unlinked upon initial removal.
* We perform sweeps by the thread hitting threshold (rather than
* background threads or by spreading work to other threads)
* because in the main contexts in which removal occurs, the
* caller is timed-out or cancelled, which are not time-critical
* enough to warrant the overhead that alternatives would impose
* on other threads.
*
* Because the sweepVotes estimate is conservative, and because
* nodes become unlinked "naturally" as they fall off the head of
* the queue, and because we allow votes to accumulate even while
* sweeps are in progress, there are typically significantly fewer
* such nodes than estimated. Choice of a threshold value
* balances the likelihood of wasted effort and contention, versus
* providing a worst-case bound on retention of interior nodes in
* quiescent queues. The value defined below was chosen
* empirically to balance these under various timeout scenarios.
*
* Because traversal operations on the linked list of nodes are a
* natural opportunity to sweep dead nodes, we generally do so,
* including all the operations that might remove elements as they
* traverse, such as removeIf and Iterator.remove. This largely
* eliminates long chains of dead interior nodes, except from
* cancelled or timed out blocking operations.
*
* Note that we cannot self-link unlinked interior nodes during
* sweeps. However, the associated garbage chains terminate when
* some successor ultimately falls off the head of the list and is
* self-linked.
*/
/** True if on multiprocessor */
private static final boolean MP =
Runtime.getRuntime().availableProcessors() > 1;
/**
* The number of times to spin (with randomly interspersed calls
* to Thread.yield) on multiprocessor before blocking when a node
* is apparently the first waiter in the queue. See above for
* explanation. Must be a power of two. The value is empirically
* derived -- it works pretty well across a variety of processors,
* numbers of CPUs, and OSes.
*/
private static final int FRONT_SPINS = 1 << 7;
/**
* The number of times to spin before blocking when a node is
* preceded by another node that is apparently spinning. Also
* serves as an increment to FRONT_SPINS on phase changes, and as
* base average frequency for yielding during spins. Must be a
* power of two.
*/
private static final int CHAINED_SPINS = FRONT_SPINS >>> 1;
/**
* The maximum number of estimated removal failures (sweepVotes)
* to tolerate before sweeping through the queue unlinking
* cancelled nodes that were not unlinked upon initial
* removal. See above for explanation. The value must be at least
* two to avoid useless sweeps when removing trailing nodes.
*/
static final int SWEEP_THRESHOLD = 32;
/**
* Queue nodes. Uses Object, not E, for items to allow forgetting
* them after use. Writes that are intrinsically ordered wrt
* other accesses or CASes use simple relaxed forms.
*/
static final class Node {
final boolean isData; // false if this is a request node
volatile Object item; // initially non-null if isData; CASed to match
volatile Node next;
volatile Thread waiter; // null when not waiting for a match
/**
* Constructs a data node holding item if item is non-null,
* else a request node. Uses relaxed write because item can
* only be seen after piggy-backing publication via CAS.
*/
Node(Object item) {
ITEM.set(this, item);
isData = (item != null);
}
/** Constructs a (matched data) dummy node. */
Node() {
isData = true;
}
final boolean casNext(Node cmp, Node val) {
// assert val != null;
return NEXT.compareAndSet(this, cmp, val);
}
final boolean casItem(Object cmp, Object val) {
// assert isData == (cmp != null);
// assert isData == (val == null);
// assert !(cmp instanceof Node);
return ITEM.compareAndSet(this, cmp, val);
}
/**
* Links node to itself to avoid garbage retention. Called
* only after CASing head field, so uses relaxed write.
*/
final void selfLink() {
// assert isMatched();
NEXT.setRelease(this, this);
}
final void appendRelaxed(Node next) {
// assert next != null;
// assert this.next == null;
NEXT.set(this, next);
}
/**
* Sets item (of a request node) to self and waiter to null,
* to avoid garbage retention after matching or cancelling.
* Uses relaxed writes because order is already constrained in
* the only calling contexts: item is forgotten only after
* volatile/atomic mechanics that extract items, and visitors
* of request nodes only ever check whether item is null.
* Similarly, clearing waiter follows either CAS or return
* from park (if ever parked; else we don't care).
*/
final void forgetContents() {
// assert isMatched();
if (!isData)
ITEM.set(this, this);
WAITER.set(this, null);
}
/**
* Returns true if this node has been matched, including the
* case of artificial matches due to cancellation.
*/
final boolean isMatched() {
return isData == (item == null);
}
/** Tries to CAS-match this node; if successful, wakes waiter. */
final boolean tryMatch(Object cmp, Object val) {
if (casItem(cmp, val)) {
LockSupport.unpark(waiter);
return true;
}
return false;
}
/**
* Returns true if a node with the given mode cannot be
* appended to this node because this node is unmatched and
* has opposite data mode.
*/
final boolean cannotPrecede(boolean haveData) {
boolean d = isData;
return d != haveData && d != (item == null);
}
private static final long serialVersionUID = -3375979862319811754L;
}
/**
* A node from which the first live (non-matched) node (if any)
* can be reached in O(1) time.
* Invariants:
* - all live nodes are reachable from head via .next
* - head != null
* - (tmp = head).next != tmp || tmp != head
* Non-invariants:
* - head may or may not be live
* - it is permitted for tail to lag behind head, that is, for tail
* to not be reachable from head!
*/
transient volatile Node head;
/**
* A node from which the last node on list (that is, the unique
* node with node.next == null) can be reached in O(1) time.
* Invariants:
* - the last node is always reachable from tail via .next
* - tail != null
* Non-invariants:
* - tail may or may not be live
* - it is permitted for tail to lag behind head, that is, for tail
* to not be reachable from head!
* - tail.next may or may not be self-linked.
*/
private transient volatile Node tail;
/** The number of apparent failures to unsplice cancelled nodes */
private transient volatile int sweepVotes;
private boolean casTail(Node cmp, Node val) {
// assert cmp != null;
// assert val != null;
return TAIL.compareAndSet(this, cmp, val);
}
private boolean casHead(Node cmp, Node val) {
return HEAD.compareAndSet(this, cmp, val);
}
/** Atomic version of ++sweepVotes. */
private int incSweepVotes() {
return (int) SWEEPVOTES.getAndAdd(this, 1) + 1;
}
/**
* Tries to CAS pred.next (or head, if pred is null) from c to p.
* Caller must ensure that we're not unlinking the trailing node.
*/
private boolean tryCasSuccessor(Node pred, Node c, Node p) {
// assert p != null;
// assert c.isData != (c.item != null);
// assert c != p;
if (pred != null)
return pred.casNext(c, p);
if (casHead(c, p)) {
c.selfLink();
return true;
}
return false;
}
/**
* Collapses dead (matched) nodes between pred and q.
* @param pred the last known live node, or null if none
* @param c the first dead node
* @param p the last dead node
* @param q p.next: the next live node, or null if at end
* @return pred if pred still alive and CAS succeeded; else p
*/
private Node skipDeadNodes(Node pred, Node c, Node p, Node q) {
// assert pred != c;
// assert p != q;
// assert c.isMatched();
// assert p.isMatched();
if (q == null) {
// Never unlink trailing node.
if (c == p) return pred;
q = p;
}
return (tryCasSuccessor(pred, c, q)
&& (pred == null || !pred.isMatched()))
? pred : p;
}
/**
* Collapses dead (matched) nodes from h (which was once head) to p.
* Caller ensures all nodes from h up to and including p are dead.
*/
private void skipDeadNodesNearHead(Node h, Node p) {
// assert h != null;
// assert h != p;
// assert p.isMatched();
for (;;) {
final Node q;
if ((q = p.next) == null) break;
else if (!q.isMatched()) { p = q; break; }
else if (p == (p = q)) return;
}
if (casHead(h, p))
h.selfLink();
}
/* Possible values for "how" argument in xfer method. */
private static final int NOW = 0; // for untimed poll, tryTransfer
private static final int ASYNC = 1; // for offer, put, add
private static final int SYNC = 2; // for transfer, take
private static final int TIMED = 3; // for timed poll, tryTransfer
/**
* Implements all queuing methods. See above for explanation.
*
* @param e the item or null for take
* @param haveData true if this is a put, else a take
* @param how NOW, ASYNC, SYNC, or TIMED
* @param nanos timeout in nanosecs, used only if mode is TIMED
* @return an item if matched, else e
* @throws NullPointerException if haveData mode but e is null
*/
@SuppressWarnings("unchecked")
private E xfer(E e, boolean haveData, int how, long nanos) {
if (haveData && (e == null))
throw new NullPointerException();
restart: for (Node s = null, t = null, h = null;;) {
for (Node p = (t != (t = tail) && t.isData == haveData) ? t
: (h = head);; ) {
final Node q; final Object item;
if (p.isData != haveData
&& haveData == ((item = p.item) == null)) {
if (h == null) h = head;
if (p.tryMatch(item, e)) {
if (h != p) skipDeadNodesNearHead(h, p);
return (E) item;
}
}
if ((q = p.next) == null) {
if (how == NOW) return e;
if (s == null) s = new Node(e);
if (!p.casNext(null, s)) continue;
if (p != t) casTail(t, s);
if (how == ASYNC) return e;
return awaitMatch(s, p, e, (how == TIMED), nanos);
}
if (p == (p = q)) continue restart;
}
}
}
/**
* Spins/yields/blocks until node s is matched or caller gives up.
*
* @param s the waiting node
* @param pred the predecessor of s, or null if unknown (the null
* case does not occur in any current calls but may in possible
* future extensions)
* @param e the comparison value for checking match
* @param timed if true, wait only until timeout elapses
* @param nanos timeout in nanosecs, used only if timed is true
* @return matched item, or e if unmatched on interrupt or timeout
*/
private E awaitMatch(Node s, Node pred, E e, boolean timed, long nanos) {
final long deadline = timed ? System.nanoTime() + nanos : 0L;
Thread w = Thread.currentThread();
int spins = -1; // initialized after first item and cancel checks
ThreadLocalRandom randomYields = null; // bound if needed
for (;;) {
final Object item;
if ((item = s.item) != e) { // matched
// assert item != s;
s.forgetContents(); // avoid garbage
@SuppressWarnings("unchecked") E itemE = (E) item;
return itemE;
}
else if (w.isInterrupted() || (timed && nanos <= 0L)) {
// try to cancel and unlink
if (s.casItem(e, s.isData ? null : s)) {
unsplice(pred, s);
return e;
}
// return normally if lost CAS
}
else if (spins < 0) { // establish spins at/near front
if ((spins = spinsFor(pred, s.isData)) > 0)
randomYields = ThreadLocalRandom.current();
}
else if (spins > 0) { // spin
--spins;
if (randomYields.nextInt(CHAINED_SPINS) == 0)
Thread.yield(); // occasionally yield
}
else if (s.waiter == null) {
s.waiter = w; // request unpark then recheck
}
else if (timed) {
nanos = deadline - System.nanoTime();
if (nanos > 0L)
LockSupport.parkNanos(this, nanos);
}
else {
LockSupport.park(this);
}
}
}
/**
* Returns spin/yield value for a node with given predecessor and
* data mode. See above for explanation.
*/
private static int spinsFor(Node pred, boolean haveData) {
if (MP && pred != null) {
if (pred.isData != haveData) // phase change
return FRONT_SPINS + CHAINED_SPINS;
if (pred.isMatched()) // probably at front
return FRONT_SPINS;
if (pred.waiter == null) // pred apparently spinning
return CHAINED_SPINS;
}
return 0;
}
/* -------------- Traversal methods -------------- */
/**
* Returns the first unmatched data node, or null if none.
* Callers must recheck if the returned node is unmatched
* before using.
*/
final Node firstDataNode() {
Node first = null;
restartFromHead: for (;;) {
Node h = head, p = h;
while (p != null) {
if (p.item != null) {
if (p.isData) {
first = p;
break;
}
}
else if (!p.isData)
break;
final Node q;
if ((q = p.next) == null)
break;
if (p == (p = q))
continue restartFromHead;
}
if (p != h && casHead(h, p))
h.selfLink();
return first;
}
}
/**
* Traverses and counts unmatched nodes of the given mode.
* Used by methods size and getWaitingConsumerCount.
*/
private int countOfMode(boolean data) {
restartFromHead: for (;;) {
int count = 0;
for (Node p = head; p != null;) {
if (!p.isMatched()) {
if (p.isData != data)
return 0;
if (++count == Integer.MAX_VALUE)
break; // @see Collection.size()
}
if (p == (p = p.next))
continue restartFromHead;
}
return count;
}
}
public String toString() {
String[] a = null;
restartFromHead: for (;;) {
int charLength = 0;
int size = 0;
for (Node p = head; p != null;) {
Object item = p.item;
if (p.isData) {
if (item != null) {
if (a == null)
a = new String[4];
else if (size == a.length)
a = Arrays.copyOf(a, 2 * size);
String s = item.toString();
a[size++] = s;
charLength += s.length();
}
} else if (item == null)
break;
if (p == (p = p.next))
continue restartFromHead;
}
if (size == 0)
return "[]";
return Helpers.toString(a, size, charLength);
}
}
private Object[] toArrayInternal(Object[] a) {
Object[] x = a;
restartFromHead: for (;;) {
int size = 0;
for (Node p = head; p != null;) {
Object item = p.item;
if (p.isData) {
if (item != null) {
if (x == null)
x = new Object[4];
else if (size == x.length)
x = Arrays.copyOf(x, 2 * (size + 4));
x[size++] = item;
}
} else if (item == null)
break;
if (p == (p = p.next))
continue restartFromHead;
}
if (x == null)
return new Object[0];
else if (a != null && size <= a.length) {
if (a != x)
System.arraycopy(x, 0, a, 0, size);
if (size < a.length)
a[size] = null;
return a;
}
return (size == x.length) ? x : Arrays.copyOf(x, size);
}
}
/**
* Returns an array containing all of the elements in this queue, in
* proper sequence.
*
* <p>The returned array will be "safe" in that no references to it are
* maintained by this queue. (In other words, this method must allocate
* a new array). The caller is thus free to modify the returned array.
*
* <p>This method acts as bridge between array-based and collection-based
* APIs.
*
* @return an array containing all of the elements in this queue
*/
public Object[] toArray() {
return toArrayInternal(null);
}
/**
* Returns an array containing all of the elements in this queue, in
* proper sequence; the runtime type of the returned array is that of
* the specified array. If the queue fits in the specified array, it
* is returned therein. Otherwise, a new array is allocated with the
* runtime type of the specified array and the size of this queue.
*
* <p>If this queue fits in the specified array with room to spare
* (i.e., the array has more elements than this queue), the element in
* the array immediately following the end of the queue is set to
* {@code null}.
*
* <p>Like the {@link #toArray()} method, this method acts as bridge between
* array-based and collection-based APIs. Further, this method allows
* precise control over the runtime type of the output array, and may,
* under certain circumstances, be used to save allocation costs.
*
* <p>Suppose {@code x} is a queue known to contain only strings.
* The following code can be used to dump the queue into a newly
* allocated array of {@code String}:
*
* <pre> {@code String[] y = x.toArray(new String[0]);}</pre>
*
* Note that {@code toArray(new Object[0])} is identical in function to
* {@code toArray()}.
*
* @param a the array into which the elements of the queue are to
* be stored, if it is big enough; otherwise, a new array of the
* same runtime type is allocated for this purpose
* @return an array containing all of the elements in this queue
* @throws ArrayStoreException if the runtime type of the specified array
* is not a supertype of the runtime type of every element in
* this queue
* @throws NullPointerException if the specified array is null
*/
@SuppressWarnings("unchecked")
public <T> T[] toArray(T[] a) {
Objects.requireNonNull(a);
return (T[]) toArrayInternal(a);
}
/**
* Weakly-consistent iterator.
*
* Lazily updated ancestor is expected to be amortized O(1) remove(),
* but O(n) in the worst case, when lastRet is concurrently deleted.
*/
final class Itr implements Iterator<E> {
private Node nextNode; // next node to return item for
private E nextItem; // the corresponding item
private Node lastRet; // last returned node, to support remove
private Node ancestor; // Helps unlink lastRet on remove()
/**
* Moves to next node after pred, or first node if pred null.
*/
@SuppressWarnings("unchecked")
private void advance(Node pred) {
for (Node p = (pred == null) ? head : pred.next, c = p;
p != null; ) {
final Object item;
if ((item = p.item) != null && p.isData) {
nextNode = p;
nextItem = (E) item;
if (c != p)
tryCasSuccessor(pred, c, p);
return;
}
else if (!p.isData && item == null)
break;
if (c != p && !tryCasSuccessor(pred, c, c = p)) {
pred = p;
c = p = p.next;
}
else if (p == (p = p.next)) {
pred = null;
c = p = head;
}
}
nextItem = null;
nextNode = null;
}
Itr() {
advance(null);
}
public final boolean hasNext() {
return nextNode != null;
}
public final E next() {
final Node p;
if ((p = nextNode) == null) throw new NoSuchElementException();
E e = nextItem;
advance(lastRet = p);
return e;
}
public void forEachRemaining(Consumer<? super E> action) {
Objects.requireNonNull(action);
Node q = null;
for (Node p; (p = nextNode) != null; advance(q = p))
action.accept(nextItem);
if (q != null)
lastRet = q;
}
public final void remove() {
final Node lastRet = this.lastRet;
if (lastRet == null)
throw new IllegalStateException();
this.lastRet = null;
if (lastRet.item == null) // already deleted?
return;
// Advance ancestor, collapsing intervening dead nodes
Node pred = ancestor;
for (Node p = (pred == null) ? head : pred.next, c = p, q;
p != null; ) {
if (p == lastRet) {
final Object item;
if ((item = p.item) != null)
p.tryMatch(item, null);
if ((q = p.next) == null) q = p;
if (c != q) tryCasSuccessor(pred, c, q);
ancestor = pred;
return;
}
final Object item; final boolean pAlive;
if (pAlive = ((item = p.item) != null && p.isData)) {
// exceptionally, nothing to do
}
else if (!p.isData && item == null)
break;
if ((c != p && !tryCasSuccessor(pred, c, c = p)) || pAlive) {
pred = p;
c = p = p.next;
}
else if (p == (p = p.next)) {
pred = null;
c = p = head;
}
}
// traversal failed to find lastRet; must have been deleted;
// leave ancestor at original location to avoid overshoot;
// better luck next time!
// assert lastRet.isMatched();
}
}
/** A customized variant of Spliterators.IteratorSpliterator */
final class LTQSpliterator implements Spliterator<E> {
static final int MAX_BATCH = 1 << 25; // max batch array size;
Node current; // current node; null until initialized
int batch; // batch size for splits
boolean exhausted; // true when no more nodes
LTQSpliterator() {}
public Spliterator<E> trySplit() {
Node p, q;
if ((p = current()) == null || (q = p.next) == null)
return null;
int i = 0, n = batch = Math.min(batch + 1, MAX_BATCH);
Object[] a = null;
do {
final Object item = p.item;
if (p.isData) {
if (item != null) {
if (a == null)
a = new Object[n];
a[i++] = item;
}
} else if (item == null) {
p = null;
break;
}
if (p == (p = q))
p = firstDataNode();
} while (p != null && (q = p.next) != null && i < n);
setCurrent(p);
return (i == 0) ? null :
Spliterators.spliterator(a, 0, i, (Spliterator.ORDERED |
Spliterator.NONNULL |
Spliterator.CONCURRENT));
}
public void forEachRemaining(Consumer<? super E> action) {
Objects.requireNonNull(action);
final Node p;
if ((p = current()) != null) {
current = null;
exhausted = true;
forEachFrom(action, p);
}
}
@SuppressWarnings("unchecked")
public boolean tryAdvance(Consumer<? super E> action) {
Objects.requireNonNull(action);
Node p;
if ((p = current()) != null) {
E e = null;
do {
final Object item = p.item;
final boolean isData = p.isData;
if (p == (p = p.next))
p = head;
if (isData) {
if (item != null) {
e = (E) item;
break;
}
}
else if (item == null)
p = null;
} while (p != null);
setCurrent(p);
if (e != null) {
action.accept(e);
return true;
}
}
return false;
}
private void setCurrent(Node p) {
if ((current = p) == null)
exhausted = true;
}
private Node current() {
Node p;
if ((p = current) == null && !exhausted)
setCurrent(p = firstDataNode());
return p;
}
public long estimateSize() { return Long.MAX_VALUE; }
public int characteristics() {
return (Spliterator.ORDERED |
Spliterator.NONNULL |
Spliterator.CONCURRENT);
}
}
/**
* Returns a {@link Spliterator} over the elements in this queue.
*
* <p>The returned spliterator is
* <a href="package-summary.html#Weakly"><i>weakly consistent</i></a>.
*
* <p>The {@code Spliterator} reports {@link Spliterator#CONCURRENT},
* {@link Spliterator#ORDERED}, and {@link Spliterator#NONNULL}.
*
* @implNote
* The {@code Spliterator} implements {@code trySplit} to permit limited
* parallelism.
*
* @return a {@code Spliterator} over the elements in this queue
* @since 1.8
*/
public Spliterator<E> spliterator() {
return new LTQSpliterator();
}
/* -------------- Removal methods -------------- */
/**
* Unsplices (now or later) the given deleted/cancelled node with
* the given predecessor.
*
* @param pred a node that was at one time known to be the
* predecessor of s
* @param s the node to be unspliced
*/
final void unsplice(Node pred, Node s) {
// assert pred != null;
// assert pred != s;
// assert s != null;
// assert s.isMatched();
// assert (SWEEP_THRESHOLD & (SWEEP_THRESHOLD - 1)) == 0;
s.waiter = null; // disable signals
/*
* See above for rationale. Briefly: if pred still points to
* s, try to unlink s. If s cannot be unlinked, because it is
* trailing node or pred might be unlinked, and neither pred
* nor s are head or offlist, add to sweepVotes, and if enough
* votes have accumulated, sweep.
*/
if (pred != null && pred.next == s) {
Node n = s.next;
if (n == null ||
(n != s && pred.casNext(s, n) && pred.isMatched())) {
for (;;) { // check if at, or could be, head
Node h = head;
if (h == pred || h == s)
return; // at head or list empty
if (!h.isMatched())
break;
Node hn = h.next;
if (hn == null)
return; // now empty
if (hn != h && casHead(h, hn))
h.selfLink(); // advance head
}
// sweep every SWEEP_THRESHOLD votes
if (pred.next != pred && s.next != s // recheck if offlist
&& (incSweepVotes() & (SWEEP_THRESHOLD - 1)) == 0)
sweep();
}
}
}
/**
* Unlinks matched (typically cancelled) nodes encountered in a
* traversal from head.
*/
private void sweep() {
for (Node p = head, s, n; p != null && (s = p.next) != null; ) {
if (!s.isMatched())
// Unmatched nodes are never self-linked
p = s;
else if ((n = s.next) == null) // trailing node is pinned
break;
else if (s == n) // stale
// No need to also check for p == s, since that implies s == n
p = head;
else
p.casNext(s, n);
}
}
/**
* Creates an initially empty {@code LinkedTransferQueue}.
*/
public LinkedTransferQueue() {
head = tail = new Node();
}
/**
* Creates a {@code LinkedTransferQueue}
* initially containing the elements of the given collection,
* added in traversal order of the collection's iterator.
*
* @param c the collection of elements to initially contain
* @throws NullPointerException if the specified collection or any
* of its elements are null
*/
public LinkedTransferQueue(Collection<? extends E> c) {
Node h = null, t = null;
for (E e : c) {
Node newNode = new Node(Objects.requireNonNull(e));
if (h == null)
h = t = newNode;
else
t.appendRelaxed(t = newNode);
}
if (h == null)
h = t = new Node();
head = h;
tail = t;
}
/**
* Inserts the specified element at the tail of this queue.
* As the queue is unbounded, this method will never block.
*
* @throws NullPointerException if the specified element is null
*/
public void put(E e) {
xfer(e, true, ASYNC, 0);
}
/**
* Inserts the specified element at the tail of this queue.
* As the queue is unbounded, this method will never block or
* return {@code false}.
*
* @return {@code true} (as specified by
* {@link BlockingQueue#offer(Object,long,TimeUnit) BlockingQueue.offer})
* @throws NullPointerException if the specified element is null
*/
public boolean offer(E e, long timeout, TimeUnit unit) {
xfer(e, true, ASYNC, 0);
return true;
}
/**
* Inserts the specified element at the tail of this queue.
* As the queue is unbounded, this method will never return {@code false}.
*
* @return {@code true} (as specified by {@link Queue#offer})
* @throws NullPointerException if the specified element is null
*/
public boolean offer(E e) {
xfer(e, true, ASYNC, 0);
return true;
}
/**
* Inserts the specified element at the tail of this queue.
* As the queue is unbounded, this method will never throw
* {@link IllegalStateException} or return {@code false}.
*
* @return {@code true} (as specified by {@link Collection#add})
* @throws NullPointerException if the specified element is null
*/
public boolean add(E e) {
xfer(e, true, ASYNC, 0);
return true;
}
/**
* Transfers the element to a waiting consumer immediately, if possible.
*
* <p>More precisely, transfers the specified element immediately
* if there exists a consumer already waiting to receive it (in
* {@link #take} or timed {@link #poll(long,TimeUnit) poll}),
* otherwise returning {@code false} without enqueuing the element.
*
* @throws NullPointerException if the specified element is null
*/
public boolean tryTransfer(E e) {
return xfer(e, true, NOW, 0) == null;
}
/**
* Transfers the element to a consumer, waiting if necessary to do so.
*
* <p>More precisely, transfers the specified element immediately
* if there exists a consumer already waiting to receive it (in
* {@link #take} or timed {@link #poll(long,TimeUnit) poll}),
* else inserts the specified element at the tail of this queue
* and waits until the element is received by a consumer.
*
* @throws NullPointerException if the specified element is null
*/
public void transfer(E e) throws InterruptedException {
if (xfer(e, true, SYNC, 0) != null) {
Thread.interrupted(); // failure possible only due to interrupt
throw new InterruptedException();
}
}
/**
* Transfers the element to a consumer if it is possible to do so
* before the timeout elapses.
*
* <p>More precisely, transfers the specified element immediately
* if there exists a consumer already waiting to receive it (in
* {@link #take} or timed {@link #poll(long,TimeUnit) poll}),
* else inserts the specified element at the tail of this queue
* and waits until the element is received by a consumer,
* returning {@code false} if the specified wait time elapses
* before the element can be transferred.
*
* @throws NullPointerException if the specified element is null
*/
public boolean tryTransfer(E e, long timeout, TimeUnit unit)
throws InterruptedException {
if (xfer(e, true, TIMED, unit.toNanos(timeout)) == null)
return true;
if (!Thread.interrupted())
return false;
throw new InterruptedException();
}
public E take() throws InterruptedException {
E e = xfer(null, false, SYNC, 0);
if (e != null)
return e;
Thread.interrupted();
throw new InterruptedException();
}
public E poll(long timeout, TimeUnit unit) throws InterruptedException {
E e = xfer(null, false, TIMED, unit.toNanos(timeout));
if (e != null || !Thread.interrupted())
return e;
throw new InterruptedException();
}
public E poll() {
return xfer(null, false, NOW, 0);
}
/**
* @throws NullPointerException {@inheritDoc}
* @throws IllegalArgumentException {@inheritDoc}
*/
public int drainTo(Collection<? super E> c) {
Objects.requireNonNull(c);
if (c == this)
throw new IllegalArgumentException();
int n = 0;
for (E e; (e = poll()) != null; n++)
c.add(e);
return n;
}
/**
* @throws NullPointerException {@inheritDoc}
* @throws IllegalArgumentException {@inheritDoc}
*/
public int drainTo(Collection<? super E> c, int maxElements) {
Objects.requireNonNull(c);
if (c == this)
throw new IllegalArgumentException();
int n = 0;
for (E e; n < maxElements && (e = poll()) != null; n++)
c.add(e);
return n;
}
/**
* Returns an iterator over the elements in this queue in proper sequence.
* The elements will be returned in order from first (head) to last (tail).
*
* <p>The returned iterator is
* <a href="package-summary.html#Weakly"><i>weakly consistent</i></a>.
*
* @return an iterator over the elements in this queue in proper sequence
*/
public Iterator<E> iterator() {
return new Itr();
}
public E peek() {
restartFromHead: for (;;) {
for (Node p = head; p != null;) {
Object item = p.item;
if (p.isData) {
if (item != null) {
@SuppressWarnings("unchecked") E e = (E) item;
return e;
}
}
else if (item == null)
break;
if (p == (p = p.next))
continue restartFromHead;
}
return null;
}
}
/**
* Returns {@code true} if this queue contains no elements.
*
* @return {@code true} if this queue contains no elements
*/
public boolean isEmpty() {
return firstDataNode() == null;
}
public boolean hasWaitingConsumer() {
restartFromHead: for (;;) {
for (Node p = head; p != null;) {
Object item = p.item;
if (p.isData) {
if (item != null)
break;
}
else if (item == null)
return true;
if (p == (p = p.next))
continue restartFromHead;
}
return false;
}
}
/**
* Returns the number of elements in this queue. If this queue
* contains more than {@code Integer.MAX_VALUE} elements, returns
* {@code Integer.MAX_VALUE}.
*
* <p>Beware that, unlike in most collections, this method is
* <em>NOT</em> a constant-time operation. Because of the
* asynchronous nature of these queues, determining the current
* number of elements requires an O(n) traversal.
*
* @return the number of elements in this queue
*/
public int size() {
return countOfMode(true);
}
public int getWaitingConsumerCount() {
return countOfMode(false);
}
/**
* Removes a single instance of the specified element from this queue,
* if it is present. More formally, removes an element {@code e} such
* that {@code o.equals(e)}, if this queue contains one or more such
* elements.
* Returns {@code true} if this queue contained the specified element
* (or equivalently, if this queue changed as a result of the call).
*
* @param o element to be removed from this queue, if present
* @return {@code true} if this queue changed as a result of the call
*/
public boolean remove(Object o) {
if (o == null) return false;
restartFromHead: for (;;) {
for (Node p = head, pred = null; p != null; ) {
Node q = p.next;
final Object item;
if ((item = p.item) != null) {
if (p.isData) {
if (o.equals(item) && p.tryMatch(item, null)) {
skipDeadNodes(pred, p, p, q);
return true;
}
pred = p; p = q; continue;
}
}
else if (!p.isData)
break;
for (Node c = p;; q = p.next) {
if (q == null || !q.isMatched()) {
pred = skipDeadNodes(pred, c, p, q); p = q; break;
}
if (p == (p = q)) continue restartFromHead;
}
}
return false;
}
}
/**
* Returns {@code true} if this queue contains the specified element.
* More formally, returns {@code true} if and only if this queue contains
* at least one element {@code e} such that {@code o.equals(e)}.
*
* @param o object to be checked for containment in this queue
* @return {@code true} if this queue contains the specified element
*/
public boolean contains(Object o) {
if (o == null) return false;
restartFromHead: for (;;) {
for (Node p = head, pred = null; p != null; ) {
Node q = p.next;
final Object item;
if ((item = p.item) != null) {
if (p.isData) {
if (o.equals(item))
return true;
pred = p; p = q; continue;
}
}
else if (!p.isData)
break;
for (Node c = p;; q = p.next) {
if (q == null || !q.isMatched()) {
pred = skipDeadNodes(pred, c, p, q); p = q; break;
}
if (p == (p = q)) continue restartFromHead;
}
}
return false;
}
}
/**
* Always returns {@code Integer.MAX_VALUE} because a
* {@code LinkedTransferQueue} is not capacity constrained.
*
* @return {@code Integer.MAX_VALUE} (as specified by
* {@link BlockingQueue#remainingCapacity()})
*/
public int remainingCapacity() {
return Integer.MAX_VALUE;
}
/**
* Saves this queue to a stream (that is, serializes it).
*
* @param s the stream
* @throws java.io.IOException if an I/O error occurs
* @serialData All of the elements (each an {@code E}) in
* the proper order, followed by a null
*/
private void writeObject(java.io.ObjectOutputStream s)
throws java.io.IOException {
s.defaultWriteObject();
for (E e : this)
s.writeObject(e);
// Use trailing null as sentinel
s.writeObject(null);
}
/**
* Reconstitutes this queue from a stream (that is, deserializes it).
* @param s the stream
* @throws ClassNotFoundException if the class of a serialized object
* could not be found
* @throws java.io.IOException if an I/O error occurs
*/
private void readObject(java.io.ObjectInputStream s)
throws java.io.IOException, ClassNotFoundException {
// Read in elements until trailing null sentinel found
Node h = null, t = null;
for (Object item; (item = s.readObject()) != null; ) {
Node newNode = new Node(item);
if (h == null)
h = t = newNode;
else
t.appendRelaxed(t = newNode);
}
if (h == null)
h = t = new Node();
head = h;
tail = t;
}
/**
* @throws NullPointerException {@inheritDoc}
*/
public boolean removeIf(Predicate<? super E> filter) {
Objects.requireNonNull(filter);
return bulkRemove(filter);
}
/**
* @throws NullPointerException {@inheritDoc}
*/
public boolean removeAll(Collection<?> c) {
Objects.requireNonNull(c);
return bulkRemove(e -> c.contains(e));
}
/**
* @throws NullPointerException {@inheritDoc}
*/
public boolean retainAll(Collection<?> c) {
Objects.requireNonNull(c);
return bulkRemove(e -> !c.contains(e));
}
public void clear() {
bulkRemove(e -> true);
}
/**
* Tolerate this many consecutive dead nodes before CAS-collapsing.
* Amortized cost of clear() is (1 + 1/MAX_HOPS) CASes per element.
*/
private static final int MAX_HOPS = 8;
/** Implementation of bulk remove methods. */
@SuppressWarnings("unchecked")
private boolean bulkRemove(Predicate<? super E> filter) {
boolean removed = false;
restartFromHead: for (;;) {
int hops = MAX_HOPS;
// c will be CASed to collapse intervening dead nodes between
// pred (or head if null) and p.
for (Node p = head, c = p, pred = null, q; p != null; p = q) {
q = p.next;
final Object item; boolean pAlive;
if (pAlive = ((item = p.item) != null && p.isData)) {
if (filter.test((E) item)) {
if (p.tryMatch(item, null))
removed = true;
pAlive = false;
}
}
else if (!p.isData && item == null)
break;
if (pAlive || q == null || --hops == 0) {
// p might already be self-linked here, but if so:
// - CASing head will surely fail
// - CASing pred's next will be useless but harmless.
if ((c != p && !tryCasSuccessor(pred, c, c = p))
|| pAlive) {
// if CAS failed or alive, abandon old pred
hops = MAX_HOPS;
pred = p;
c = q;
}
} else if (p == q)
continue restartFromHead;
}
return removed;
}
}
/**
* Runs action on each element found during a traversal starting at p.
* If p is null, the action is not run.
*/
@SuppressWarnings("unchecked")
void forEachFrom(Consumer<? super E> action, Node p) {
for (Node pred = null; p != null; ) {
Node q = p.next;
final Object item;
if ((item = p.item) != null) {
if (p.isData) {
action.accept((E) item);
pred = p; p = q; continue;
}
}
else if (!p.isData)
break;
for (Node c = p;; q = p.next) {
if (q == null || !q.isMatched()) {
pred = skipDeadNodes(pred, c, p, q); p = q; break;
}
if (p == (p = q)) { pred = null; p = head; break; }
}
}
}
/**
* @throws NullPointerException {@inheritDoc}
*/
public void forEach(Consumer<? super E> action) {
Objects.requireNonNull(action);
forEachFrom(action, head);
}
// VarHandle mechanics
private static final VarHandle HEAD;
private static final VarHandle TAIL;
private static final VarHandle SWEEPVOTES;
static final VarHandle ITEM;
static final VarHandle NEXT;
static final VarHandle WAITER;
static {
try {
MethodHandles.Lookup l = MethodHandles.lookup();
HEAD = l.findVarHandle(LinkedTransferQueue.class, "head",
Node.class);
TAIL = l.findVarHandle(LinkedTransferQueue.class, "tail",
Node.class);
SWEEPVOTES = l.findVarHandle(LinkedTransferQueue.class, "sweepVotes",
int.class);
ITEM = l.findVarHandle(Node.class, "item", Object.class);
NEXT = l.findVarHandle(Node.class, "next", Node.class);
WAITER = l.findVarHandle(Node.class, "waiter", Thread.class);
} catch (ReflectiveOperationException e) {
throw new ExceptionInInitializerError(e);
}
// Reduce the risk of rare disastrous classloading in first call to
// LockSupport.park: https://bugs.openjdk.java.net/browse/JDK-8074773
Class<?> ensureLoaded = LockSupport.class;
}
}
Loading...
Report
Report success
We will send you the feedback within 2 working days through the letter!
Please fill in the reason for the report carefully. Provide as detailed a description as possible.
Please select a report type
Cancel
Send
误判申诉

此处可能存在不合适展示的内容,页面不予展示。您可通过相关编辑功能自查并修改。

如您确认内容无涉及 不当用语 / 纯广告导流 / 暴力 / 低俗色情 / 侵权 / 盗版 / 虚假 / 无价值内容或违法国家有关法律法规的内容,可点击提交进行申诉,我们将尽快为您处理。

取消
提交

Releases

No release

Contributors

All

Language(Optional)

Activities

can not load any more
Edit
About
Homepage
马建仓 AI 助手
尝试更多
代码解读
代码找茬
代码优化
1
https://gitee.com/gwdcode/jdk11.0.2lib.src.java.base.java.git
git@gitee.com:gwdcode/jdk11.0.2lib.src.java.base.java.git
gwdcode
jdk11.0.2lib.src.java.base.java
JDK11.0.2-lib.src.java.base.java
master
Going to Help Center

Search

Comment
Repository Report
Back to the top
Login prompt
This operation requires login to the code cloud account. Please log in before operating.
Go to login
No account. Register

AltStyle によって変換されたページ (->オリジナル) /