- Install dependencies with
pip install -r requirements.txt
- Fill up the queue by running
python tasks.py 500
- Start Celery with
celery -A tasks worker --loglevel=INFO -c 3
- Disconnect computer from internet until you receive a
CANCEL_TASKS_BY_DEFAULT
warning. This may take a couple minutes and you'll see several errors, but you have to wait until the warning. - Reconnect to the internet. After a couple seconds, Celery will reconnect and tasks will start flowing again.
- Let tasks run for about a minute. You'll notice a slight slowdown and then suddenly tasks will stop entirely for a couple minutes, before resuming again.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
# Mount the data disk | |
# Adapted from https://gist.github.com/ctompkinson/30f570882af38879b36fc7bffe3d1a60 | |
device=/dev/sdh | |
mount_point=/monitor-config | |
apt-get install -y nvme-cli | |
while [ true ]; do | |
for blkdev in $(nvme list | awk '/^\/dev/ { print $1 }'); do | |
mapping=$(nvme id-ctrl --raw-binary "${blkdev}" | cut -c3073-3104 | tr -s ' ' | sed 's/ $//g') | |
if [ ${mapping} = ${device} ]; then | |
echo "Found $device on $blkdev" |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import timeit | |
from abc import ABC, abstractmethod | |
class A(ABC): | |
@abstractmethod | |
def a(self): | |
... | |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import json | |
from dataclasses import dataclass, is_dataclass, asdict | |
class DataclassEncoder(json.JSONEncoder): | |
def default(self, obj): | |
if is_dataclass(obj): | |
return asdict(obj) | |
return super().default(obj) |
OlderNewer