mirror of
https://github.com/yacy/yacy_search_server.git
synced 2024-09-21 00:00:13 +02:00
6491270b3a
removed preferred IPv4 in start options and added a new field IP6 in peer seeds which will contain one or more IPv6 addresses. Now every peer has one or more IP addresses assigned, even several IPv6 addresses are possible. The peer-ping process must check all given and possible IP addresses for a backping and return the one IP which was successful when pinging the peer. The ping-ing peer must be able to recognize which of the given IPs are available for outside access of the peer and store this accordingly. If only one IPv6 address is available and no IPv4, then the IPv6 is stored in the old IP field of the seed DNA. Many methods in Seed.java are now marked as @deprecated because they had been used for a single IP only. There is still a large construction site left in YaCy now where all these deprecated methods must be replaced with new method calls. The 'extra'-IPs, used by cluster assignment had been removed since that can be replaced with IPv6 usage in p2p clusters. All clusters must now use IPv6 if they want an intranet-routing.
294 lines
13 KiB
Java
294 lines
13 KiB
Java
// Transmission.java
|
|
// (C) 2009 by Michael Peter Christen; mc@yacy.net, Frankfurt a. M., Germany
|
|
// first published 09.02.2009 on http://yacy.net
|
|
//
|
|
// $LastChangedDate$
|
|
// $LastChangedRevision$
|
|
// $LastChangedBy$
|
|
//
|
|
// LICENSE
|
|
//
|
|
// This program is free software; you can redistribute it and/or modify
|
|
// it under the terms of the GNU General Public License as published by
|
|
// the Free Software Foundation; either version 2 of the License, or
|
|
// (at your option) any later version.
|
|
//
|
|
// This program is distributed in the hope that it will be useful,
|
|
// but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
// GNU General Public License for more details.
|
|
//
|
|
// You should have received a copy of the GNU General Public License
|
|
// along with this program; if not, write to the Free Software
|
|
// Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
|
|
|
|
package net.yacy.peers;
|
|
|
|
import java.io.IOException;
|
|
import java.util.ArrayList;
|
|
import java.util.HashSet;
|
|
import java.util.Iterator;
|
|
import java.util.List;
|
|
import java.util.Random;
|
|
import java.util.Set;
|
|
|
|
import net.yacy.cora.document.encoding.ASCII;
|
|
import net.yacy.cora.storage.HandleSet;
|
|
import net.yacy.cora.util.ConcurrentLog;
|
|
import net.yacy.cora.util.SpaceExceededException;
|
|
import net.yacy.kelondro.data.word.Word;
|
|
import net.yacy.kelondro.data.word.WordReference;
|
|
import net.yacy.kelondro.data.word.WordReferenceRow;
|
|
import net.yacy.kelondro.index.RowHandleSet;
|
|
import net.yacy.kelondro.rwi.ReferenceContainer;
|
|
import net.yacy.kelondro.rwi.ReferenceContainerCache;
|
|
import net.yacy.kelondro.workflow.WorkflowJob;
|
|
import net.yacy.search.index.Segment;
|
|
|
|
public class Transmission {
|
|
|
|
// The number of RWIs we can be sure a remote peer will accept
|
|
// anything beyond that might get discarded without notice
|
|
public static final int maxRWIsCount = 1000; // since SVN 7993 hardcoded in htroot/yacy/transferRWI.java:161
|
|
|
|
protected ConcurrentLog log;
|
|
protected Segment segment;
|
|
protected SeedDB seeds;
|
|
protected boolean gzipBody4Transfer;
|
|
protected int timeout4Transfer;
|
|
|
|
public Transmission(
|
|
final ConcurrentLog log,
|
|
final Segment segment,
|
|
final SeedDB seeds,
|
|
final boolean gzipBody4Transfer,
|
|
final int timeout4Transfer) {
|
|
this.log = log;
|
|
this.segment = segment;
|
|
this.seeds = seeds;
|
|
this.gzipBody4Transfer = gzipBody4Transfer;
|
|
this.timeout4Transfer = timeout4Transfer;
|
|
}
|
|
|
|
public Chunk newChunk(final Seed dhtTarget) {
|
|
return new Chunk(dhtTarget);
|
|
}
|
|
|
|
public class Chunk extends WorkflowJob implements Iterable<ReferenceContainer<WordReference>> {
|
|
/**
|
|
* a dispatcher entry contains
|
|
* - the primary target, which is a word hash, as marker for the entry
|
|
* - a set of indexContainers in a cache,
|
|
* - the associated URLs in a set to have a cache for the transmission
|
|
* to multiple peers and to ensure that all entries in the indexContainers
|
|
* have a reference in the urls
|
|
* - a set of yacy seeds which will shrink as the containers are transmitted to them
|
|
* - a counter that gives the number of sucessful and unsuccessful transmissions so far
|
|
*/
|
|
private final Seed dhtTarget;
|
|
private final ReferenceContainerCache<WordReference> containers;
|
|
private final HandleSet references;
|
|
private final HandleSet badReferences;
|
|
private int hit, miss;
|
|
|
|
/**
|
|
* generate a new dispatcher target. such a target is defined with a primary target and
|
|
* a set of target peers that shall receive the entries of the containers
|
|
* the payloadrow defines the structure of container entries
|
|
* @param dhtTarget
|
|
*/
|
|
public Chunk(final Seed dhtTarget) {
|
|
super();
|
|
this.dhtTarget = dhtTarget;
|
|
this.containers = new ReferenceContainerCache<WordReference>(Segment.wordReferenceFactory, Segment.wordOrder, Word.commonHashLength);
|
|
this.references = new RowHandleSet(WordReferenceRow.urlEntryRow.primaryKeyLength, WordReferenceRow.urlEntryRow.objectOrder, 0);
|
|
this.badReferences = new RowHandleSet(WordReferenceRow.urlEntryRow.primaryKeyLength, WordReferenceRow.urlEntryRow.objectOrder, 0);
|
|
this.hit = 0;
|
|
this.miss = 0;
|
|
}
|
|
|
|
/*
|
|
* return a new container with at most max elements and put the rest back to the index
|
|
* as this chunk might be transferred back to myself a random selection needs to be taken
|
|
* @param container
|
|
* @param max
|
|
* @throws RowSpaceExceededException
|
|
* @return
|
|
*/
|
|
private ReferenceContainer<WordReference> trimContainer(final ReferenceContainer<WordReference> container, final int max) throws SpaceExceededException {
|
|
final ReferenceContainer<WordReference> c = new ReferenceContainer<WordReference>(Segment.wordReferenceFactory, container.getTermHash(), max);
|
|
final int part = container.size() / max + 1;
|
|
final Random r = new Random();
|
|
WordReference w;
|
|
final List<byte[]> selected = new ArrayList<byte[]>();
|
|
final Iterator<WordReference> i = container.entries();
|
|
while ((i.hasNext()) && (c.size() < max)) {
|
|
w = i.next();
|
|
if (r.nextInt(part) == 0) {
|
|
c.add(w);
|
|
selected.add(w.urlhash());
|
|
}
|
|
}
|
|
// remove the selected entries from container
|
|
for (final byte[] b : selected) container.removeReference(b);
|
|
// put container back
|
|
try {
|
|
Transmission.this.segment.storeRWI(container);
|
|
} catch (final Exception e) {
|
|
ConcurrentLog.logException(e);
|
|
}
|
|
return c;
|
|
}
|
|
|
|
/**
|
|
* add a container to the Entry cache.
|
|
* all entries in the container are checked and only such are stored which have a reference entry
|
|
* @param container
|
|
* @throws SpaceExceededException
|
|
*/
|
|
public void add(final ReferenceContainer<WordReference> container) throws SpaceExceededException {
|
|
int remaining = maxRWIsCount;
|
|
for (final ReferenceContainer<WordReference> ic : this) remaining -= ic.size();
|
|
if (remaining <= 0) {
|
|
// No space left in this chunk
|
|
try {
|
|
Transmission.this.segment.storeRWI(container);
|
|
} catch (final Exception e) {
|
|
ConcurrentLog.logException(e);
|
|
}
|
|
return;
|
|
}
|
|
final ReferenceContainer<WordReference> c = (remaining >= container.size()) ? container : trimContainer(container, remaining);
|
|
// iterate through the entries in the container and check if the reference is in the repository
|
|
final List<byte[]> notFoundx = new ArrayList<byte[]>();
|
|
Set<String> testids = new HashSet<String>();
|
|
Iterator<WordReference> i = c.entries();
|
|
while (i.hasNext()) {
|
|
final WordReference e = i.next();
|
|
if (this.references.has(e.urlhash())) continue;
|
|
if (this.badReferences.has(e.urlhash())) {
|
|
notFoundx.add(e.urlhash());
|
|
continue;
|
|
}
|
|
testids.add(ASCII.String(e.urlhash()));
|
|
}
|
|
i = c.entries();
|
|
while (i.hasNext()) {
|
|
final WordReference e = i.next();
|
|
try {
|
|
if (Transmission.this.segment.fulltext().getLoadTime(ASCII.String(e.urlhash())) >= 0) {
|
|
this.references.put(e.urlhash());
|
|
} else {
|
|
notFoundx.add(e.urlhash());
|
|
this.badReferences.put(e.urlhash());
|
|
}
|
|
} catch (IOException e1) {
|
|
ConcurrentLog.logException(e1);
|
|
notFoundx.add(e.urlhash());
|
|
this.badReferences.put(e.urlhash());
|
|
}
|
|
}
|
|
// now delete all references that were not found
|
|
for (final byte[] b : notFoundx) c.removeReference(b);
|
|
// finally add the remaining container to the cache
|
|
this.containers.add(c);
|
|
}
|
|
|
|
/**
|
|
* get all containers from the entry. This method may be used to flush remaining entries
|
|
* if they had been finished transmission without success (not enough peers arrived)
|
|
*/
|
|
@Override
|
|
public Iterator<ReferenceContainer<WordReference>> iterator() {
|
|
return this.containers.iterator();
|
|
}
|
|
|
|
public int containersSize() {
|
|
return this.containers.size();
|
|
}
|
|
|
|
public Seed dhtTarget() {
|
|
return this.dhtTarget;
|
|
}
|
|
|
|
/**
|
|
* return the number of successful transmissions
|
|
* @return
|
|
*/
|
|
public int hit() {
|
|
return this.hit;
|
|
}
|
|
|
|
/**
|
|
* return the number of unsuccessful transmissions
|
|
* @return
|
|
*/
|
|
public int miss() {
|
|
return this.miss;
|
|
}
|
|
|
|
public boolean transmit() {
|
|
// transferring selected words to remote peer
|
|
if (this.dhtTarget == Transmission.this.seeds.mySeed() || this.dhtTarget.hash.equals(Transmission.this.seeds.mySeed().hash)) {
|
|
// target is my own peer. This is easy. Just restore the indexContainer
|
|
restore();
|
|
this.hit++;
|
|
Transmission.this.log.info("Transfer of chunk to myself-target");
|
|
return true;
|
|
}
|
|
Transmission.this.log.info("starting new index transmission request to " + this.dhtTarget.getName());
|
|
final long start = System.currentTimeMillis();
|
|
final String error = Protocol.transferIndex(Transmission.this.seeds, this.dhtTarget, this.containers, this.references, Transmission.this.segment, Transmission.this.gzipBody4Transfer, Transmission.this.timeout4Transfer);
|
|
if (error == null) {
|
|
// words successfully transfered
|
|
final long transferTime = System.currentTimeMillis() - start;
|
|
final Iterator<ReferenceContainer<WordReference>> i = this.containers.iterator();
|
|
final ReferenceContainer<WordReference> firstContainer = (i == null) ? null : i.next();
|
|
Transmission.this.log.info("Index transfer of " + this.containers.size() +
|
|
" references for terms [ " + ((firstContainer == null) ? null : ASCII.String(firstContainer.getTermHash())) + " ..]" +
|
|
" and " + this.references.size() + " URLs" +
|
|
" to peer " + this.dhtTarget.getName() + ":" + this.dhtTarget.hash +
|
|
" in " + (transferTime / 1000) +
|
|
" seconds successful (" + (1000 * this.containers.size() / (transferTime + 1)) +
|
|
" words/s)");
|
|
Transmission.this.seeds.mySeed().incSI(this.containers.size());
|
|
Transmission.this.seeds.mySeed().incSU(this.references.size());
|
|
// if the peer has set a pause time and we are in flush mode (index transfer)
|
|
// then we pause for a while now
|
|
Transmission.this.log.info("Transfer finished of chunk to target " + this.dhtTarget.hash + "/" + this.dhtTarget.getName());
|
|
this.hit++;
|
|
return true;
|
|
}
|
|
Transmission.this.log.info(
|
|
"Index transfer to peer " + this.dhtTarget.getName() + ":" + this.dhtTarget.hash +
|
|
" failed: " + error);
|
|
this.miss++;
|
|
// write information that peer does not receive index transmissions
|
|
Transmission.this.log.info("Transfer failed of chunk to target " + this.dhtTarget.hash + "/" + this.dhtTarget.getName() + ": " + error);
|
|
// get possibly newer target Info
|
|
final Seed newTarget = Transmission.this.seeds.get(this.dhtTarget.hash);
|
|
if (newTarget != null) {
|
|
if (this.dhtTarget.clash(newTarget.getIPs())) {
|
|
newTarget.setFlagAcceptRemoteIndex(false);
|
|
Transmission.this.seeds.update(newTarget.hash, newTarget);
|
|
} else {
|
|
// we tried an old Address. Don't change anything
|
|
}
|
|
} else {
|
|
// target not in DB anymore. ???
|
|
}
|
|
return false;
|
|
}
|
|
|
|
public void restore() {
|
|
for (final ReferenceContainer<WordReference> ic : this) try {
|
|
Transmission.this.segment.storeRWI(ic);
|
|
} catch (final Exception e) {
|
|
ConcurrentLog.logException(e);
|
|
}
|
|
}
|
|
|
|
}
|
|
|
|
}
|