Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Improve offer publishing #4950

Merged
merged 7 commits into from
Dec 29, 2020
Merged

Improve offer publishing #4950

Show file tree
Hide file tree
Changes from 6 commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
b553130
Cleanup
chimp1984 Dec 15, 2020
7bfbd0f
Change handling of delay at republishing
chimp1984 Dec 15, 2020
8e71575
Add else case for check for offer is in source list and deActivate state
chimp1984 Dec 16, 2020
24f53e0
Add flush method.
chimp1984 Dec 16, 2020
1049b7d
Remove delay for publishing
chimp1984 Dec 16, 2020
83e620d
Fix missing param in test
chimp1984 Dec 16, 2020
c5d0fa7
Move stopPeriodicRefreshOffersTimer before processListForRepublishOffers
chimp1984 Dec 23, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
86 changes: 51 additions & 35 deletions core/src/main/java/bisq/core/offer/OpenOfferManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,6 +49,7 @@
import bisq.network.p2p.NodeAddress;
import bisq.network.p2p.P2PService;
import bisq.network.p2p.SendDirectMessageListener;
import bisq.network.p2p.peers.Broadcaster;
import bisq.network.p2p.peers.PeerManager;

import bisq.common.Timer;
Expand Down Expand Up @@ -117,6 +118,7 @@ public class OpenOfferManager implements PeerManager.Listener, DecryptedDirectMe
private final RefundAgentManager refundAgentManager;
private final DaoFacade daoFacade;
private final FilterManager filterManager;
private final Broadcaster broadcaster;
private final PersistenceManager<TradableList<OpenOffer>> persistenceManager;
private final Map<String, OpenOffer> offersToBeEdited = new HashMap<>();
private final TradableList<OpenOffer> openOffers = new TradableList<>();
Expand Down Expand Up @@ -148,6 +150,7 @@ public OpenOfferManager(CreateOfferService createOfferService,
RefundAgentManager refundAgentManager,
DaoFacade daoFacade,
FilterManager filterManager,
Broadcaster broadcaster,
PersistenceManager<TradableList<OpenOffer>> persistenceManager) {
this.createOfferService = createOfferService;
this.keyRing = keyRing;
Expand All @@ -166,6 +169,7 @@ public OpenOfferManager(CreateOfferService createOfferService,
this.refundAgentManager = refundAgentManager;
this.daoFacade = daoFacade;
this.filterManager = filterManager;
this.broadcaster = broadcaster;
this.persistenceManager = persistenceManager;

this.persistenceManager.initialize(openOffers, "OpenOffers", PersistenceManager.Source.PRIVATE);
Expand Down Expand Up @@ -214,10 +218,6 @@ private void cleanUpAddressEntries() {
});
}

private void shutDown() {
shutDown(null);
}

public void shutDown(@Nullable Runnable completeHandler) {
stopped = true;
p2PService.getPeerManager().removeListener(this);
Expand All @@ -235,6 +235,11 @@ public void shutDown(@Nullable Runnable completeHandler) {
UserThread.execute(() -> openOffers.forEach(
openOffer -> offerBookService.removeOfferAtShutDown(openOffer.getOffer().getOfferPayload())
));

// Force broadcaster to send out immediately, otherwise we could have a 2 sec delay until the
// bundled messages sent out.
broadcaster.flush();

if (completeHandler != null) {
// For typical number of offers we are tolerant with delay to give enough time to broadcast.
// If number of offers is very high we limit to 3 sec. to not delay other shutdown routines.
Expand Down Expand Up @@ -871,41 +876,53 @@ private void maybeUpdatePersistedOffers() {
///////////////////////////////////////////////////////////////////////////////////////////

private void republishOffers() {
int size = openOffers.size();
final ArrayList<OpenOffer> openOffersList = new ArrayList<>(openOffers.getList());
if (!stopped) {
stopPeriodicRefreshOffersTimer();
for (int i = 0; i < size; i++) {
// we delay to avoid reaching throttle limits

long delay = 700;
final long minDelay = (i + 1) * delay;
final long maxDelay = (i + 2) * delay;
final OpenOffer openOffer = openOffersList.get(i);
UserThread.runAfterRandomDelay(() -> {
if (openOffers.contains(openOffer)) {
String id = openOffer.getId();
if (id != null && !openOffer.isDeactivated())
republishOffer(openOffer);
}
if (stopped) {
return;
}

}, minDelay, maxDelay, TimeUnit.MILLISECONDS);
}
List<OpenOffer> openOffersList = new ArrayList<>(openOffers.getList());
processListForRepublishOffers(openOffersList);

stopPeriodicRefreshOffersTimer();
}

private void processListForRepublishOffers(List<OpenOffer> list) {
if (list.isEmpty()) {
return;
}

OpenOffer openOffer = list.remove(0);
if (openOffers.contains(openOffer) && !openOffer.isDeactivated()) {
// TODO It is not clear yet if it is better for the node and the network to send out all add offer
// messages in one go or to spread it over a delay. With power users who have 100-200 offers that can have
// some significant impact to user experience and the network
republishOffer(openOffer, () -> processListForRepublishOffers(list));

/* republishOffer(openOffer,
() -> UserThread.runAfter(() -> processListForRepublishOffers(list),
30, TimeUnit.MILLISECONDS));*/
} else {
log.debug("We have stopped already. We ignore that republishOffers call.");
// If the offer was removed in the meantime or if its deactivated we skip and call
// processListForRepublishOffers again with the list where we removed the offer already.
processListForRepublishOffers(list);
}
}

private void republishOffer(OpenOffer openOffer) {
republishOffer(openOffer, null);
}

private void republishOffer(OpenOffer openOffer, @Nullable Runnable completeHandler) {
offerBookService.addOffer(openOffer.getOffer(),
() -> {
if (!stopped) {
log.debug("Successfully added offer to P2P network.");
// Refresh means we send only the data needed to refresh the TTL (hash, signature and sequence no.)
if (periodicRefreshOffersTimer == null)
if (periodicRefreshOffersTimer == null) {
startPeriodicRefreshOffersTimer();
} else {
log.debug("We have stopped already. We ignore that offerBookService.republishOffers.onSuccess call.");
}
if (completeHandler != null) {
completeHandler.run();
}
}
},
errorMessage -> {
Expand All @@ -914,26 +931,25 @@ private void republishOffer(OpenOffer openOffer) {
stopRetryRepublishOffersTimer();
retryRepublishOffersTimer = UserThread.runAfter(OpenOfferManager.this::republishOffers,
RETRY_REPUBLISH_DELAY_SEC);
} else {
log.debug("We have stopped already. We ignore that offerBookService.republishOffers.onFault call.");

if (completeHandler != null) {
completeHandler.run();
}
}
});
}

private void startPeriodicRepublishOffersTimer() {
stopped = false;
if (periodicRepublishOffersTimer == null)
if (periodicRepublishOffersTimer == null) {
periodicRepublishOffersTimer = UserThread.runPeriodically(() -> {
if (!stopped) {
republishOffers();
} else {
log.debug("We have stopped already. We ignore that periodicRepublishOffersTimer.run call.");
}
},
REPUBLISH_INTERVAL_MS,
TimeUnit.MILLISECONDS);
else
log.trace("periodicRepublishOffersTimer already stated");
}
}

private void startPeriodicRefreshOffersTimer() {
Expand Down
6 changes: 3 additions & 3 deletions core/src/test/java/bisq/core/offer/OpenOfferManagerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,7 +49,7 @@ public void testStartEditOfferForActiveOffer() {
final OpenOfferManager manager = new OpenOfferManager(null, null, null, p2PService,
null, null, null, offerBookService,
null, null, null,
null, null, null, null, null, null,
null, null, null, null, null, null, null,
persistenceManager);

AtomicBoolean startEditOfferSuccessful = new AtomicBoolean(false);
Expand Down Expand Up @@ -81,7 +81,7 @@ public void testStartEditOfferForDeactivatedOffer() {
final OpenOfferManager manager = new OpenOfferManager(null, null, null, p2PService,
null, null, null, offerBookService,
null, null, null,
null, null, null, null, null, null,
null, null, null, null, null, null, null,
persistenceManager);

AtomicBoolean startEditOfferSuccessful = new AtomicBoolean(false);
Expand All @@ -106,7 +106,7 @@ public void testStartEditOfferForOfferThatIsCurrentlyEdited() {
final OpenOfferManager manager = new OpenOfferManager(null, null, null, p2PService,
null, null, null, offerBookService,
null, null, null,
null, null, null, null, null, null,
null, null, null, null, null, null, null,
persistenceManager);

AtomicBoolean startEditOfferSuccessful = new AtomicBoolean(false);
Expand Down
4 changes: 4 additions & 0 deletions p2p/src/main/java/bisq/network/p2p/peers/Broadcaster.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -73,6 +73,10 @@ public void shutDown(Runnable resultHandler) {
}
}

public void flush() {
maybeBroadcastBundle();
}

private void doShutDown() {
broadcastHandlers.forEach(BroadcastHandler::cancel);
if (timer != null) {
Expand Down