| 549 | # Update content.json on peers |
| 550 | @util.Noparallel() |
| 551 | def publish(self, limit="default", inner_path="content.json", diffs={}, cb_progress=None): |
| 552 | published = [] # Successfully published (Peer) |
| 553 | publishers = [] # Publisher threads |
| 554 | |
| 555 | if not self.peers: |
| 556 | self.announce() |
| 557 | |
| 558 | if limit == "default": |
| 559 | limit = 5 |
| 560 | threads = limit |
| 561 | |
| 562 | peers = self.getConnectedPeers() |
| 563 | num_connected_peers = len(peers) |
| 564 | |
| 565 | random.shuffle(peers) |
| 566 | peers = sorted(peers, key=lambda peer: peer.connection.handshake.get("rev", 0) < config.rev - 100) # Prefer newer clients |
| 567 | |
| 568 | if len(peers) < limit * 2 and len(self.peers) > len(peers): # Add more, non-connected peers if necessary |
| 569 | peers += self.getRecentPeers(limit * 2) |
| 570 | |
| 571 | peers = set(peers) |
| 572 | |
| 573 | self.log.info("Publishing %s to %s/%s peers (connected: %s) diffs: %s (%.2fk)..." % ( |
| 574 | inner_path, limit, len(self.peers), num_connected_peers, list(diffs.keys()), float(len(str(diffs))) / 1024 |
| 575 | )) |
| 576 | |
| 577 | if not peers: |
| 578 | return 0 # No peers found |
| 579 | |
| 580 | event_done = gevent.event.AsyncResult() |
| 581 | for i in range(min(len(peers), limit, threads)): |
| 582 | publisher = gevent.spawn(self.publisher, inner_path, peers, published, limit, diffs, event_done, cb_progress) |
| 583 | publishers.append(publisher) |
| 584 | |
| 585 | event_done.get() # Wait for done |
| 586 | if len(published) < min(len(self.peers), limit): |
| 587 | time.sleep(0.2) # If less than we need sleep a bit |
| 588 | if len(published) == 0: |
| 589 | gevent.joinall(publishers) # No successful publish, wait for all publisher |
| 590 | |
| 591 | # Publish more peers in the backgroup |
| 592 | self.log.info( |
| 593 | "Published %s to %s peers, publishing to %s more peers in the background" % |
| 594 | (inner_path, len(published), limit) |
| 595 | ) |
| 596 | |
| 597 | for thread in range(2): |
| 598 | gevent.spawn(self.publisher, inner_path, peers, published, limit=limit * 2, diffs=diffs) |
| 599 | |
| 600 | # Send my hashfield to every connected peer if changed |
| 601 | gevent.spawn(self.sendMyHashfield, 100) |
| 602 | |
| 603 | return len(published) |
| 604 | |
| 605 | # Copy this site |
| 606 | def clone(self, address, privatekey=None, address_index=None, root_inner_path="", overwrite=False): |