-
Notifications
You must be signed in to change notification settings - Fork 20
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.
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
Allow users to check ping of instances? #12
Comments
That's a great idea, but not something that can be done within a GitHub |
No probs, feel free to close out, I just thought I'd suggest it as I went through this recently with two instances and the difference was night and day. |
Could we add to the tables the hosting country/region of each instance server perhaps? |
@calculuschild would you prefer the emoji flag of the country or the 2-digit ISO abbreviation? |
Either way. I'd probably prefer the flag because it's easier to distinguish at a glance. As long as it's clear it's the hosting region, not the language. |
I'm also interested in this. It could have an argument to be used manually. This script calculates the average latency for a list of domains using the The Example of script usage:
This will calculate the average latency for each domain in the import argparse
import subprocess
from typing import List
DOMAINS = ["example.com", "example.org", "example.net"]
NUM_PINGS = 10
DELAY = 60
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser()
parser.add_argument("--latency", action="store_true", help="Calculate average latency for domains")
return parser.parse_args()
def ping_domain(domain: str) -> float:
result = subprocess.run(["ping", "-c", str(NUM_PINGS), domain], capture_output=True, text=True)
lines = result.stdout.strip().split("\n")
avg_line = [line for line in lines if "avg" in line][0]
avg_time = float(avg_line.split("/")[-3])
return avg_time
def run_latency_check(args: argparse.Namespace):
command_args = [f"{domain} {NUM_PINGS}" for domain in DOMAINS]
command_str = "python3 -c 'from my_module import ping_domain; domain, num_pings = \"{arg}\".split(); print(ping_domain(domain, int(num_pings)))'"
subprocess.run(["parallel", "-j0", f"--delay {DELAY}", command_str, ":::", *command_args])
def main():
args = parse_args()
if args.latency:
run_latency_check(args)
if __name__ == "__main__":
main() |
#!/usr/bin/env python3
import json
import asyncio
import aiohttp
import time
from typing import List, Dict
TIME_BETWEEN_REQUESTS = 5
TIME_TOTAL = 60
async def get_latency(session, domain):
try:
start = time.time()
if not domain.startswith(("http://", "https://")):
domain = "https://" + domain
async with session.get(domain, timeout=3) as response:
end = time.time()
return end - start
except asyncio.TimeoutError:
return float("inf")
except aiohttp.client_exceptions.ServerDisconnectedError:
return float("inf")
def add_latency_to_domain(domain, latency, latencies):
if domain not in latencies:
latencies[domain] = []
latencies[domain].append(latency)
return latencies
async def measure_latencies_for_domains(session, domains, latencies):
tasks = []
for domain in domains:
tasks.append(get_latency(session, domain))
results = await asyncio.gather(*tasks)
for domain, latency in zip(domains, results):
latencies = add_latency_to_domain(domain, latency, latencies)
return latencies
async def measure_latencies(domains, duration):
latencies = {}
start_time = time.time()
end_time = start_time + duration
async with aiohttp.ClientSession() as session:
while time.time() < end_time:
latencies = await measure_latencies_for_domains(session, domains, latencies)
await asyncio.sleep(TIME_BETWEEN_REQUESTS)
return latencies
def average_latencies(latencies):
averages = []
for domain, latency_list in latencies.items():
avg_latency = sum(latency_list) / len(latency_list)
averages.append((domain, avg_latency))
return averages
def sort_latencies(averages):
return sorted(averages, key=lambda x: x[1])
async def get_latency_report(domains, duration):
latencies = await measure_latencies(domains, duration)
averages = average_latencies(latencies)
return sort_latencies(averages)
def get_instances(data: Dict) -> List[Dict]:
instances = []
for instance_details in data["instance_details"]:
instances.append(instance_details)
return instances
def get_domains(instances: List[Dict]) -> List[str]:
return [instance["domain"] for instance in instances]
def load_json_data(filepath: str) -> Dict:
with open(filepath) as json_data:
return json.load(json_data)
async def main():
data = load_json_data('stats.json')
instances = get_instances(data)
domains = get_domains(instances)
report = await get_latency_report(domains, TIME_TOTAL)
for domain, avg_latency in report:
print(f"{domain}: {avg_latency:.2f} seconds")
if __name__ == "__main__":
parser = argparse.ArgumentParser()
parser.add_argument('--latency', action='store_true', help='Execute latency measurement')
args = parser.parse_args()
if args.latency:
asyncio.run(main()) |
One of the best things new users can do is find a Lemmy server with a low ping from their location, as this will make the whole experience faster and more enjoyable for them.
It would be great to encourage users to find a server near them, perhaps by providing a "Check ping" button for each listing which performs a client-side ping to the instance and shows the result.
The text was updated successfully, but these errors were encountered: