distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Peer profile organizer — classifies and selects peers.
2
3Ported from net.i2p.router.peermanager.ProfileOrganizer.
4"""
5
6from __future__ import annotations
7
8import random
9import time
10from enum import Enum
11from typing import TYPE_CHECKING
12
13if TYPE_CHECKING:
14 from i2p_peer.profile import PeerProfile
15
16
17class PeerTier(Enum):
18 FAST = "fast" # High speed + high capacity
19 HIGH_CAPACITY = "high_cap" # High capacity, any speed
20 STANDARD = "standard" # Meets minimum thresholds
21 FAILING = "failing" # Below minimum thresholds
22 BANNED = "banned" # Explicitly banned
23
24
25# Tier ordering for comparison (lower index = better tier)
26_TIER_ORDER = [PeerTier.FAST, PeerTier.HIGH_CAPACITY, PeerTier.STANDARD,
27 PeerTier.FAILING, PeerTier.BANNED]
28
29
30class ProfileOrganizer:
31 """Classifies peers into tiers and selects peers for tunnel building.
32
33 Tier thresholds (configurable):
34 - FAST: capacity >= 0.8 AND speed >= 0.7
35 - HIGH_CAPACITY: capacity >= 0.6
36 - STANDARD: capacity >= 0.3
37 - FAILING: capacity < 0.3
38 - BANNED: explicitly banned
39 """
40
41 # Configurable thresholds
42 FAST_CAPACITY_THRESHOLD = 0.8
43 FAST_SPEED_THRESHOLD = 0.7
44 HIGH_CAP_THRESHOLD = 0.6
45 STANDARD_THRESHOLD = 0.3
46 MIN_ESTABLISHED_FOR_FAST = 10 # Min tunnel builds before eligible for fast tier
47
48 def __init__(self):
49 self._profiles: dict[bytes, PeerProfile] = {}
50 self._tiers: dict[PeerTier, set[bytes]] = {t: set() for t in PeerTier}
51
52 def add_profile(self, profile: PeerProfile) -> None:
53 """Add or update a peer profile and reclassify."""
54 h = profile.router_hash
55 # Remove from old tier if present
56 self._remove_from_tiers(h)
57 self._profiles[h] = profile
58 tier = self.classify(profile)
59 self._tiers[tier].add(h)
60
61 def remove_profile(self, router_hash: bytes) -> None:
62 self._remove_from_tiers(router_hash)
63 self._profiles.pop(router_hash, None)
64
65 def get_profile(self, router_hash: bytes) -> PeerProfile | None:
66 return self._profiles.get(router_hash)
67
68 def classify(self, profile: PeerProfile) -> PeerTier:
69 """Determine which tier a peer belongs to."""
70 if profile.is_currently_banned:
71 return PeerTier.BANNED
72
73 total_builds = profile.tunnel_builds_succeeded + profile.tunnel_builds_failed
74
75 if (profile.capacity >= self.FAST_CAPACITY_THRESHOLD and
76 profile.speed >= self.FAST_SPEED_THRESHOLD and
77 total_builds >= self.MIN_ESTABLISHED_FOR_FAST):
78 return PeerTier.FAST
79
80 if profile.capacity >= self.HIGH_CAP_THRESHOLD:
81 return PeerTier.HIGH_CAPACITY
82
83 if profile.capacity >= self.STANDARD_THRESHOLD:
84 return PeerTier.STANDARD
85
86 return PeerTier.FAILING
87
88 def reclassify_all(self) -> None:
89 """Reclassify all peers (call periodically)."""
90 self._tiers = {t: set() for t in PeerTier}
91 for h, profile in self._profiles.items():
92 tier = self.classify(profile)
93 self._tiers[tier].add(h)
94
95 def select_fast_peers(self, count: int, exclude: set[bytes] | None = None) -> list[bytes]:
96 """Select peers from FAST tier for tunnel building."""
97 return self._select_from_tier(PeerTier.FAST, count, exclude)
98
99 def select_high_capacity_peers(self, count: int, exclude: set[bytes] | None = None) -> list[bytes]:
100 """Select from HIGH_CAPACITY tier."""
101 return self._select_from_tier(PeerTier.HIGH_CAPACITY, count, exclude)
102
103 def select_peers(self, count: int, min_tier: PeerTier = PeerTier.STANDARD,
104 exclude: set[bytes] | None = None) -> list[bytes]:
105 """Select peers at or above minimum tier."""
106 exclude = exclude or set()
107 min_idx = _TIER_ORDER.index(min_tier)
108 eligible_tiers = _TIER_ORDER[:min_idx + 1]
109 candidates = []
110 for tier in eligible_tiers:
111 for h in self._tiers[tier]:
112 if h not in exclude:
113 candidates.append(h)
114 random.shuffle(candidates)
115 return candidates[:count]
116
117 def get_tier(self, router_hash: bytes) -> PeerTier:
118 for tier, hashes in self._tiers.items():
119 if router_hash in hashes:
120 return tier
121 return PeerTier.FAILING
122
123 @property
124 def fast_count(self) -> int:
125 return len(self._tiers[PeerTier.FAST])
126
127 @property
128 def high_capacity_count(self) -> int:
129 return len(self._tiers[PeerTier.HIGH_CAPACITY])
130
131 @property
132 def standard_count(self) -> int:
133 return len(self._tiers[PeerTier.STANDARD])
134
135 @property
136 def failing_count(self) -> int:
137 return len(self._tiers[PeerTier.FAILING])
138
139 @property
140 def total_count(self) -> int:
141 return len(self._profiles)
142
143 def _remove_from_tiers(self, router_hash: bytes) -> None:
144 for tier_set in self._tiers.values():
145 tier_set.discard(router_hash)
146
147 def _select_from_tier(self, tier: PeerTier, count: int,
148 exclude: set[bytes] | None = None) -> list[bytes]:
149 exclude = exclude or set()
150 candidates = [h for h in self._tiers[tier] if h not in exclude]
151 random.shuffle(candidates)
152 return candidates[:count]