Page MenuHomeSoftware Heritage
Differential D7030 Diff 25485 swh/objstorage/tests/winery_benchmark.py

Changeset View

Standalone View

View Options

swh/objstorage/tests/winery_benchmark.py

# Copyright (C) 2021 The Software Heritage developers # Copyright (C) 2021-2022 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution # See the AUTHORS file at the top-level directory of this distribution
# License: GNU General Public License version 3, or any later version # License: GNU General Public License version 3, or any later version
# See top-level LICENSE file for more information # See top-level LICENSE file for more information
import asyncio import asyncio
import concurrent.futures import concurrent.futures
import logging import logging
import os import os
import random import random
import time import time
from swh.objstorage.backends.winery.stats import Stats
from swh.objstorage.factory import get_objstorage from swh.objstorage.factory import get_objstorage
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
def work(kind, args): def work(kind, args):
return Worker(args).run(kind) return Worker(args).run(kind)
class Worker(object): class Worker(Stats):
def __init__(self, args): def __init__(self, args):
super().__init__(args.get("output_dir"))
self.args = args self.args = args
vlorentzUnsubmitted
Done
Inline Actions
return Worker(args).run(kind)
- class Worker(Stats):
+ class Worker:
def __init__(self, args):
- super().__init__(args.get("output_dir"))
+ self._stats = Stats(args.get("output_dir")
self.args = args
def run(self, kind):

we prefer collaborator patterns to inheritance for this sort of stuff

vlorentz: we prefer collaborator patterns to inheritance for this sort of stuff
dacharyAuthorUnsubmitted
Done
Inline Actions

Thanks for the advice, changed accordingly.

dachary: Thanks for the advice, changed accordingly.
def run(self, kind): def run(self, kind):
getattr(self, kind)() getattr(self, kind)()
return kind return kind
def ro(self): def ro(self):
self.storage = get_objstorage( self.storage = get_objstorage(
cls="winery", cls="winery",
readonly=True, readonly=True,
base_dsn=self.args["base_dsn"], base_dsn=self.args["base_dsn"],
shard_dsn=self.args["shard_dsn"], shard_dsn=self.args["shard_dsn"],
shard_max_size=self.args["shard_max_size"], shard_max_size=self.args["shard_max_size"],
throttle_read=self.args["throttle_read"], throttle_read=self.args["throttle_read"],
throttle_write=self.args["throttle_write"], throttle_write=self.args["throttle_write"],
output_dir=self.args.get("output_dir"),
) )
with self.storage.winery.base.db.cursor() as c: with self.storage.winery.base.db.cursor() as c:
while True: while True:
c.execute( c.execute(
"SELECT signature FROM signature2shard WHERE inflight = FALSE " "SELECT signature FROM signature2shard WHERE inflight = FALSE "
"ORDER BY random() LIMIT %s", "ORDER BY random() LIMIT %s",
(self.args["ro_worker_max_request"],), (self.args["ro_worker_max_request"],),
) )
if c.rowcount > 0: if c.rowcount > 0:
break break
logger.info(f"Worker(ro, {os.getpid()}): empty, waiting") logger.info(f"Worker(ro, {os.getpid()}): empty, waiting")
time.sleep(1) time.sleep(1)
logger.info(f"Worker(ro, {os.getpid()}): requesting {c.rowcount} objects") logger.info(f"Worker(ro, {os.getpid()}): requesting {c.rowcount} objects")
start = time.time() start = time.time()
for row in c: for row in c:
obj_id = row[0].tobytes() obj_id = row[0].tobytes()
assert self.storage.get(obj_id) is not None content = self.storage.get(obj_id)
assert content is not None
if self.stats_active:
self.stats_read(obj_id, content)
elapsed = time.time() - start elapsed = time.time() - start
logger.info(f"Worker(ro, {os.getpid()}): finished ({elapsed:.2f}s)") logger.info(f"Worker(ro, {os.getpid()}): finished ({elapsed:.2f}s)")
def payloads_define(self): def payloads_define(self):
self.payloads = [ self.payloads = [
3 * 1024 + 1, 3 * 1024 + 1,
3 * 1024 + 1, 3 * 1024 + 1,
3 * 1024 + 1, 3 * 1024 + 1,
Show All 9 Linesclass Worker(Stats):
def rw(self): def rw(self):
self.storage = get_objstorage( self.storage = get_objstorage(
cls="winery", cls="winery",
base_dsn=self.args["base_dsn"], base_dsn=self.args["base_dsn"],
shard_dsn=self.args["shard_dsn"], shard_dsn=self.args["shard_dsn"],
shard_max_size=self.args["shard_max_size"], shard_max_size=self.args["shard_max_size"],
throttle_read=self.args["throttle_read"], throttle_read=self.args["throttle_read"],
throttle_write=self.args["throttle_write"], throttle_write=self.args["throttle_write"],
output_dir=self.args.get("output_dir"),
) )
self.payloads_define() self.payloads_define()
random_content = open("/dev/urandom", "rb") random_content = open("/dev/urandom", "rb")
logger.info(f"Worker(rw, {os.getpid()}): start") logger.info(f"Worker(rw, {os.getpid()}): start")
start = time.time() start = time.time()
count = 0 count = 0
while len(self.storage.winery.packers) == 0: while len(self.storage.winery.packers) == 0:
content = random_content.read(random.choice(self.payloads)) content = random_content.read(random.choice(self.payloads))
self.storage.add(content=content) obj_id = self.storage.add(content=content)
if self.stats_active:
self.stats_write(obj_id, content)
count += 1 count += 1
logger.info(f"Worker(rw, {os.getpid()}): packing {count} objects") logger.info(f"Worker(rw, {os.getpid()}): packing {count} objects")
packer = self.storage.winery.packers[0] packer = self.storage.winery.packers[0]
packer.join() packer.join()
assert packer.exitcode == 0 assert packer.exitcode == 0
elapsed = time.time() - start elapsed = time.time() - start
logger.info(f"Worker(rw, {os.getpid()}): finished ({elapsed:.2f}s)") logger.info(f"Worker(rw, {os.getpid()}): finished ({elapsed:.2f}s)")
▲ Show 20 Lines • Show All 52 Lines • Show Last 20 Lines
About · Developer information · Legal