basak

I've been using this code "in production" for a while now. It dates back to https://groups.google.com/g/python-tulip/c/J7tCcBU5TPA/m/NM7iBhhhEAAJ except that I converted it to Trio a while ago. It is intended to be lossless - if desired, you can ensure to catch all messages since you start to listen, without losing anything. Producers are blocked on send() until the slowest consumer has received the message.

Since new consumers won't receive messages from before they began to listen, the point at which a consumer "begins listening" is important. This happens when the async iterator is created - ie. when the for loop runs the implicit aiter(). If you do this as the first thing in a coroutine, you might expect all message following a nursery.start_soon() call starting that coroutine to be picked up. But in practice, the for loop won't run the implicit aiter() until some time later, and so you won't see messages sent prior to that point. To avoid this, you must all aiter() yourself and pass that in, o

basak

#!/bin/sh
set -e

# lxd-ssh
# Wrapper for using ssh/scp/etc with lxd
# Author: Robie Basak <robie.basak@canonical.com>
# Last-Update: 2017-10-11

# Instructions:
#

basak

import asyncio

class Worker:
    '''A Worker has background asyncio tasks that are cancelled on __del__'''
    def __init__(self, loop=None):
        self._loop = loop if loop else asyncio.get_event_loop()

        # This set serves to keep incomplete tasks alive as long as this class
        # instance exists.
        self._tasks = set()

basak

	.align	2
	.thumb
	.thumb_func
	.type	slp_switch, %function
slp_switch:
.LFB133:
	.file 3 "platform/switch_arm32_gcc.h"
	.loc 3 57 0
	.cfi_startproc
	@ args = 0, pretend = 0, frame = 16

basak

	.align	2
	.thumb
	.thumb_func
	.type	slp_switch, %function
slp_switch:
.LFB133:
	.file 3 "platform/switch_arm32_gcc.h"
	.loc 3 57 0
	.cfi_startproc
	@ args = 0, pretend = 0, frame = 16

basak

	.align	2
	.thumb
	.thumb_func
	.type	slp_switch, %function
slp_switch:
.LFB133:
	.file 3 "platform/switch_arm32_gcc.h"
	.loc 3 54 0
	.cfi_startproc
	@ args = 0, pretend = 0, frame = 16

basak

#!/usr/bin/python3

from random import choice
from itertools import product
from time import sleep


class GameOfLife(object):
    def __init__(self, height, width):
        self.size = (height, width)

basak

def rental_car_discount(days):
  if days < 3:
    return 0
  elif 3 <= days < 7:
    return 20
  elif days >= 7:
    return 50

def rental_car_cost(days):
  return (days * 40) - rental_car_discount(days)

basak

#!/usr/bin/python3

STARTING_SCORE = ('0', '0')

# key: previous score; value: new score if player 0 scores a point from that
POINT_LOOKUP = {
    ('0', '0'): ('15', '0'),
    ('0', '15'): ('15', '15'),
    ('0', '30'): ('15', '30'),
    ('0', '40'): ('15', '40'),

basak

#!/usr/bin/python3

# Conway's Game of Life
# Based on Jack Diederich's implementation
# http://pyvideo.org/video/880/stop-writing-classes

from itertools import chain, product

neighbour_deltas = set(list(product([-1, 0, 1], repeat=2))) - set([(0, 0)])