In D847#18210, @ardumont wrote:

Long time overdue ;)

I remember reading this a while back [1]

[1] http://docs.celeryproject.org/en/v4.0.1/whatsnew-4.0.html#lowercase-setting-names

Now that we're 4.x all the way there's no point keeping the backwards compatibility around :). That will also allow us to use the config that the celery test fixtures are shipping (although these settings turn out to not be that useful).

Closed by commit rDSCH15262bbd64ce: Switch celery settings to lowercase names (authored by olasd). · Explain WhyDec 18 2018, 5:33 PM

This revision was automatically updated to reflect the committed changes.

Revision Contents
Changeset List

Path

Size

swh/

scheduler/

celery_backend/

config.py

41 lines

Diff 2722

View Options

swh/scheduler/celery_backend/config.py

# Copyright (C) 2019 The Software Heritage developers		# Copyright (C) 2019 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

from collections import deque
from functools import partial		from functools import partial
from typing import Optional, Iterable, Dict		from typing import Dict, Iterable, List, Optional

from swh.core.utils import grouper		from swh.core.utils import grouper
from swh.model.model import Content, BaseModel		from swh.model.model import Content, BaseModel
from swh.storage import get_storage		from swh.storage import get_storage


class BufferingProxyStorage:		class BufferingProxyStorage:
"""Storage implementation in charge of accumulating objects prior to		"""Storage implementation in charge of accumulating objects prior to
Show All 30 Lines	def __init__(self, storage, min_batch_size=None):
10010241024),		10010241024),
'skipped_content': min_batch_size.get('skipped_content', 10000),		'skipped_content': min_batch_size.get('skipped_content', 10000),
'directory': min_batch_size.get('directory', 25000),		'directory': min_batch_size.get('directory', 25000),
'revision': min_batch_size.get('revision', 100000),		'revision': min_batch_size.get('revision', 100000),
'release': min_batch_size.get('release', 100000),		'release': min_batch_size.get('release', 100000),
}		}
self.object_types = [		self.object_types = [
'content', 'skipped_content', 'directory', 'revision', 'release']		'content', 'skipped_content', 'directory', 'revision', 'release']
self._objects = {k: deque() for k in self.object_types}		self._objects = {k: {} for k in self.object_types}

def __getattr__(self, key):		def __getattr__(self, key):
if key.endswith('_add'):		if key.endswith('_add'):
object_type = key.rsplit('_', 1)[0]		object_type = key.rsplit('_', 1)[0]
if object_type in self.object_types:		if object_type in self.object_types:
return partial(		return partial(
self.object_add, object_type=object_type		self.object_add, object_type=object_type,
		keys=['id'],
)		)
if key == 'storage':		if key == 'storage':
raise AttributeError(key)		raise AttributeError(key)
return getattr(self.storage, key)		return getattr(self.storage, key)

def content_add(self, content: Iterable[Content]) -> Dict:		def content_add(self, content: Iterable[Content]) -> Dict:
"""Enqueue contents to write to the storage.		"""Enqueue contents to write to the storage.

Following policies apply:		Following policies apply:

- First, check if the queue's threshold is hit.		- First, check if the queue's threshold is hit.
If it is flush content to the storage.		If it is flush content to the storage.

- If not, check if the total size of enqueued contents's		- If not, check if the total size of enqueued contents's
threshold is hit. If it is flush content to the storage.		threshold is hit. If it is flush content to the storage.

"""		"""
content = list(content)		content = list(content)
s = self.object_add(content, object_type='content')		s = self.object_add(
		content, object_type='content',
		keys=['sha1', 'sha1_git', 'sha256', 'blake2s256'])
if not s:		if not s:
q = self._objects['content']		buffer_ = self._objects['content'].values()
total_size = sum(c.length for c in q)		total_size = sum(c.length for c in buffer_)
if total_size >= self.min_batch_size['content_bytes']:		if total_size >= self.min_batch_size['content_bytes']:
return self.flush(['content'])		return self.flush(['content'])

return s		return s

		def skipped_content_add(self, content: Iterable[Content]) -> Dict:
		return self.object_add(
		content, object_type='skipped_content',
		keys=['sha1', 'sha1_git', 'sha256', 'blake2s256'])

def flush(self, object_types: Optional[Iterable[str]] = None) -> Dict:		def flush(self, object_types: Optional[Iterable[str]] = None) -> Dict:
if object_types is None:		if object_types is None:
object_types = self.object_types		object_types = self.object_types
summary = {} # type: Dict[str, Dict]		summary = {} # type: Dict[str, Dict]
for object_type in object_types:		for object_type in object_types:
q = self._objects[object_type]		buffer_ = self._objects[object_type]
for objs in grouper(q, n=self.min_batch_size[object_type]):		batches = grouper(
		buffer_.values(), n=self.min_batch_size[object_type])
		for batch in batches:
add_fn = getattr(self.storage, '%s_add' % object_type)		add_fn = getattr(self.storage, '%s_add' % object_type)
s = add_fn(objs)		s = add_fn(batch)
summary = {k: v + summary.get(k, 0)		summary = {k: v + summary.get(k, 0)
for k, v in s.items()}		for k, v in s.items()}
q.clear()		buffer_.clear()

return summary		return summary

def object_add(		def object_add(
self, objects: Iterable[BaseModel], *, object_type: str) -> Dict:		self, objects: Iterable[BaseModel], *,
		object_type: str, keys: List[str]) -> Dict:
"""Enqueue objects to write to the storage. This checks if the queue's		"""Enqueue objects to write to the storage. This checks if the queue's
threshold is hit. If it is actually write those to the storage.		threshold is hit. If it is actually write those to the storage.

"""		"""
q = self._objects[object_type]		buffer_ = self._objects[object_type]
threshold = self.min_batch_size[object_type]		threshold = self.min_batch_size[object_type]
q.extend(objects)		for obj in objects:
if len(q) >= threshold:		obj_key = tuple(getattr(obj, key) for key in keys)
		buffer_[obj_key] = obj
		if len(buffer_) >= threshold:
return self.flush()		return self.flush()
		ardumontUnsubmitted Not Done Inline Actions (unrelated to the diff) was wondering why we flush everything and not the single object_type passed here. But that could make sense to flush everything indeed. ardumont: (unrelated to the diff) was wondering why we flush everything and not the single object_type…

return {}		return {}

Switch celery settings to lowercase namesClosedPublicActions

Details

Diff Detail

Event Timeline

Revision ContentsChangeset List

Diff 2722

swh/scheduler/celery_backend/config.py

Switch celery settings to lowercase names
ClosedPublic
Actions

Revision Contents
Changeset List