for ( Map.Entry<DHTTransportContact,Object[]> entry: replies.entrySet()){
DHTTransportContact contact = entry.getKey();
HashWrapper hw = new HashWrapper( contact.getID());
SurveyContactState contact_state = survey_state.get( hw );
if ( contact_state != null ){
contact_state.updateContactDetails( contact );
}else{
contact_state = new SurveyContactState( contact );
survey_state.put( hw, contact_state );
}
contact_state.updateUseTime();
Object[] temp = entry.getValue();
List<DHTDBMapping> mappings = (List<DHTDBMapping>)temp[0];
List<byte[]> reply = (List<byte[]>)temp[1];
if ( reply == null ){
contact_state.contactFailed();
}else{
contact_state.contactOK();
if ( mappings.size() != reply.size()){
Debug.out( "Inconsistent: mappings=" + mappings.size() + ", reply=" + reply.size());
continue;
}
Iterator<DHTDBMapping> it1 = mappings.iterator();
Iterator<byte[]> it2 = reply.iterator();
while( it1.hasNext()){
DHTDBMapping mapping = it1.next();
byte[] rep = it2.next();
if ( rep == null ){
contact_state.removeMapping( mapping );
}else{
// must match against our short-key mapping for consistency
DHTDBMapping mapping_to_check = stored_values_prefix_map.get( mapping.getShortKey());
if ( mapping_to_check == null ){
// deleted
}else{
byte[] k = mapping_to_check.getKey().getBytes();
int rep_len = rep.length;
if ( rep_len < 2 || rep_len >= k.length ){
Debug.out( "Invalid rep_len: " + rep_len );
continue;
}
boolean match = true;
int offset = k.length-rep_len;
for (int i=0;i<rep_len;i++){
if ( rep[i] != k[i+offset] ){
match = false;
break;
}
}
if ( match ){
contact_state.addMapping( mapping );
}else{
contact_state.removeMapping( mapping );
}
}
}
}
Set<DHTDBMapping> contact_mappings = contact_state.getMappings();
for ( DHTDBMapping m: contact_mappings ){
int[] t = totals.get( m );
if ( t == null ){
t = new int[]{ 2 }; // one for local node + 1 for them
totals.put( m, t );
}else{
t[0]++;
}
}
}
}
for (Map.Entry<DHTDBMapping,List<DHTTransportContact>> entry: mapping_to_node_map.entrySet()){
DHTDBMapping mapping = entry.getKey();
List<DHTTransportContact> contacts = entry.getValue();
int[] t = totals.get( mapping );
int copies;
if ( t == null ){
copies = 1; // us!
}else{
copies = t[0];
}
Iterator<DHTDBValueImpl> values = mapping.getValues();
if ( values.hasNext()){
int max_replication_factor = -1;
while( values.hasNext()){
DHTDBValueImpl value = values.next();
int rf = value.getReplicationFactor();
if ( rf > max_replication_factor ){
max_replication_factor = rf;
}
}
if ( max_replication_factor == 0 ){
continue;
}
if ( max_replication_factor > router.getK()){
max_replication_factor = router.getK();
}
if ( copies < max_replication_factor ){
int required = max_replication_factor - copies;
List<SurveyContactState> potential_targets = new ArrayList<SurveyContactState>();
List<byte[]> addresses = new ArrayList<byte[]>( contacts.size());
for ( DHTTransportContact c: contacts ){
if ( c.getProtocolVersion() < DHTTransportUDP.PROTOCOL_VERSION_REPLICATION_CONTROL3 ){
continue;
}
addresses.add( c.getAddress().getAddress().getAddress());
SurveyContactState contact_state = survey_state.get( new HashWrapper( c.getID()));
if ( contact_state != null && !contact_state.testMapping( mapping )){
potential_targets.add( contact_state );
}
}
Set<HashWrapper> bad_addresses = new HashSet<HashWrapper>();
for ( byte[] a1: addresses ){
for ( byte[] a2: addresses ){
// ignore ipv6 for the moment...
if ( a1 == a2 || a1.length != a2.length || a1.length != 4 ){
continue;
}
// ignore common /16 s
if ( a1[0] == a2[0] && a1[1] == a2[1] ){
log( "/16 match on " + ByteFormatter.encodeString( a1 ) + "/" + ByteFormatter.encodeString( a2 ));
bad_addresses.add( new HashWrapper( a1 ));
bad_addresses.add( new HashWrapper( a2 ));
}
}
}
final byte[] key = mapping.getKey().getBytes();
Collections.sort(
potential_targets,
new Comparator<SurveyContactState>()
{
public int
compare(
SurveyContactState o1,
SurveyContactState o2)
{
boolean o1_bad = o1.getConsecFails() >= 2;
boolean o2_bad = o2.getConsecFails() >= 2;
if ( o1_bad == o2_bad ){
// switch from age based to closest as per Roxana's advice
if ( false ){
long res = o2.getCreationTime() - o1.getCreationTime();
if ( res < 0 ){
return( -1 );
}else if ( res > 0 ){
return( 1 );
}else{
return( 0 );
}
}else{
return(
control.computeAndCompareDistances(
o1.getContact().getID(),
o2.getContact().getID(),
key ));
}
}else{
if ( o1_bad ){
return( 1 );
}else{
return( -1 );
}
}
}
});
int avail = Math.min( required, potential_targets.size());
for (int i=0;i<avail;i++){
SurveyContactState target = potential_targets.get( i );
if ( bad_addresses.size() > 0 &&
bad_addresses.contains( new HashWrapper( target.getContact().getAddress().getAddress().getAddress()))){
// make it look like this target has the mapping as we don't want to store it there but we want to treat it as
// if it has it, effectively reducing availability but not skewing storage in favour of potentially malicious nodes
target.addMapping( mapping );