Differential D7030 Diff 25490 swh/objstorage/tests/winery_benchmark.py

Changeset View

Standalone View

View Options

swh/objstorage/tests/winery_benchmark.py

# See the AUTHORS file at the top-level directory of this distribution

# License: GNU General Public License version 3, or any later version

# See top-level LICENSE file for more information

import asyncio

import concurrent.futures

import logging

import os

import random

import time

import psycopg2

from swh.objstorage.backends.winery.stats import Stats

from swh.objstorage.factory import get_objstorage

logger = logging.getLogger(__name__)

def work(kind, args):

return Worker(args).run(kind)

class Worker(object):

class Worker:

def __init__(self, args):

self.stats = Stats(args.get("output_dir"))

self.args = args

vlorentzUnsubmitted

Done

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

Thanks for the advice, changed accordingly.

dachary: Thanks for the advice, changed accordingly.

def run(self, kind):

getattr(self, kind)()

return kind

def ro(self):

try:

self._ro()

except psycopg2.OperationalError:

# It may happen when the database is dropped, just

# conclude the read loop gracefully and move on

pass

def _ro(self):

self.storage = get_objstorage(

cls="winery",

readonly=True,

base_dsn=self.args["base_dsn"],

shard_dsn=self.args["shard_dsn"],

shard_max_size=self.args["shard_max_size"],

throttle_read=self.args["throttle_read"],

throttle_write=self.args["throttle_write"],

output_dir=self.args.get("output_dir"),

)

with self.storage.winery.base.db.cursor() as c:

while True:

c.execute(

"SELECT signature FROM signature2shard WHERE inflight = FALSE "

"ORDER BY random() LIMIT %s",

(self.args["ro_worker_max_request"],),

)

if c.rowcount > 0:

break

logger.info(f"Worker(ro, {os.getpid()}): empty, waiting")

time.sleep(1)

logger.info(f"Worker(ro, {os.getpid()}): requesting {c.rowcount} objects")

start = time.time()

for row in c:

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.stats_active:

self.stats.stats_read(obj_id, content)

elapsed = time.time() - start

logger.info(f"Worker(ro, {os.getpid()}): finished ({elapsed:.2f}s)")

def payloads_define(self):

self.payloads = [

3 * 1024 + 1,

Show All 9 Lines

class Worker:

def rw(self):

self.storage = get_objstorage(

cls="winery",

base_dsn=self.args["base_dsn"],

shard_dsn=self.args["shard_dsn"],

shard_max_size=self.args["shard_max_size"],

throttle_read=self.args["throttle_read"],

throttle_write=self.args["throttle_write"],

output_dir=self.args.get("output_dir"),

)

self.payloads_define()

random_content = open("/dev/urandom", "rb")

logger.info(f"Worker(rw, {os.getpid()}): start")

start = time.time()

count = 0

while len(self.storage.winery.packers) == 0:

content = random_content.read(random.choice(self.payloads))

self.storage.add(content=content)

obj_id = self.storage.add(content=content)

if self.stats.stats_active:

self.stats.stats_write(obj_id, content)

count += 1

logger.info(f"Worker(rw, {os.getpid()}): packing {count} objects")

packer = self.storage.winery.packers[0]

packer.join()

assert packer.exitcode == 0

elapsed = time.time() - start

logger.info(f"Worker(rw, {os.getpid()}): finished ({elapsed:.2f}s)")

▲ Show 20 Lines • Show All 52 Lines • Show Last 20 Lines