DEV Community: Ben Steadman

Reservoir Sampling

Ben Steadman — Sun, 10 Jan 2021 20:30:00 +0000

This post originally appeared on steadbytes.com

Reservoir Sampling refers to a family of algorithms for sampling a fixed number
of elements from an input of unknown length with uniform probabilities.
In this article, I aim to answer the following questions:

When/why is Reservoir Sampling useful?
How does Reservoir Sampling work?
How can Reservoir Sampling be implemented?
How can a Reservoir Sampling implementation be practically tested for correctness?

When/why is Reservoir Sampling useful?

The main benefit of Reservoir Sampling is that it provides an upper bound on memory
usage that is invariant of the input stream length. This allows sampling input streams
which are far larger in size than the available memory of a system. It is also useful in
situations where computations on a small sample are both representative of the complete
population and faster/more efficient than operating on the complete input.

In general, Reservoir Sampling is well suited to situations where data arrives
sequentially over time and/or is too large to fit entirely into memory.

For example:

Streaming data calculations.
Testing/verifying a system in production.
- Checking every input would impact performance too much.
Estimating aggregate properties of a large dataset
- Total number of records satisfying a predicate.
Type inference on semi-structured data.
- Infer the type of each column in a large CSV file without exhaustively checking every row.

How does Reservoir Sampling work?

Though many different implementations exist (more on some of these later), they all
follow the same high level approach. A reservoir (usually represented by an array) of elements is maintained whilst the input is read sequentially. New elements replace
those in the reservoir with uniform probability. When the input is exhausted, the
reservoir contains the final sample. For a sample size $n$ , number of elements read so far $t$ and reservoir $R$ :

Initialise an empty reservoir of size $n$ .
The reservoir is populated with the first $n$ elements of the input:

(x_0,x_1,\ldots,x_n)

The subsequent $t$ th elements overwrite those in the reservoir with probability $t / (n + 1)$ . For any $(t>n)(t\gt n)$ th element:

P(x_t \varepsilon R) = t / (n + 1)

In pseudo-code:

reservoir = []
t = 0 # Number of elements read so far
while not EOF:
  m = int(t * random()) # 1 <= m < t
  if m < n:
    # Make the next element a candidate for the sample, replacing one at random
    reservoir[m] = read_next_element()
  t += 1

Another formulation of this approach was proposed by Vitter¹ (more on this work
later) which calculates the gap between elements being added to the reservoir and
"jumps" through the input in chunks:

reservoir = []
t = 0
while not EOF:
  to_skip = calculate_skip(n, t)
  skip(to_skip)
  if not EOF:
    # Which element to replace
    m = int(n * random()) # 1 <= m < n
    reservoir[m] = read_next_element()
  t += to_skip + 1

Assuming that to_skip is not prohibitively expensive, this approach reduces the CPU
usage as it avoids generating a random number for every element in the input.

How can Reservoir Sampling be implemented?

As mentioned previously, there are many well known and well studied implementations of
Reservoir Sampling - each with their own advantages and disadvantages. I'll detail
several here with the intent of providing an understanding of the basic implementation
and some of the most important improvements that have been made to it. The reader is
encouraged to research further into other implementations.

Note: The following sections assume uniformly random number generation.

Algorithm R

Attributed to Alan G. Waterman by Knuth², Algorithm R represents the simplest implementation by following the basic framework outlined previously with no optimisation.

Populate the reservoir with the first $n$ elements.
For each subsequent $t$ th element, generate a random integer $\varepsilon (0, t)$ .
If $\lt n$ (e.g. $m$ is a valid index of the reservoir), replace the $m$ th element of the reservoir with the new one. Otherwise, discard the new element.

from itertools import islice
from random import randrange
from typing import Iterable

def algorithm_r(iterable: Iterable, n: int):
    iterable = iter(iterable)
    # Initialise reservoir with first n elements
    reservoir = list(islice(iterable, n))

    for t, x in enumerate(iterable, n + 1):
        m = randrange(t)
        if m < n:  # Make the next record a candidate, replacing one at random
            reservoir[m] = x

    return reservoir

Algorithm R runs in $O (N)$ time because processing a single element takes constant time (dominated by generating a random number) and the entire input ( $N$ elements) is read. The primary CPU cost of this algorithm is the random number generation.

Method 4 (Protection Reservoir)³

Suppose the entire input is available sequentially in a single file and a uniform
random number $u = U (0, 1)$ is assigned to each item $x$ :

# u: x
0.1765: 3
0.0214: 12
0.0093: 8
0.1238: 11
0.0615: 12
0.2852: 15
0.6827: 1
0.7547: 11
0.0946: 15
0.6403: 3

Let $t$ be the $n$ th smallest of the $u$ -values. The items with $u$ -values less than
or equal to $t$ now form a uniform random sample of size $n$ from the input. For $n = 5$ :

0.1766: 3
0.0214: 12 <-- 2nd smallest
0.0093: 8  <-- Smallest
0.1238: 11 <-- 5th smallest (t)
0.0615: 7  <-- 3rd smallest
0.2852: 10
0.6827: 1
0.7547: 11
0.0946: 15 <- 4th smallest
0.6403: 3

Thus forming the sample $(12, 8, 11, 7, 15)$ . Fan et al.³ use this idea to define
Method 4:

Populate the reservoir with the first $n$ elements.
Assign each reservoir element a random number $u$ and store these in the $u$ -array.
For each subsequent $t$ th element $x$ :
1. Generate a random number $u$
2. If $u$ is smaller than the largest value in the $u$ -array, replace the largest value in the $u$ -array with $u$ and replace the corresponding element of the reservoir with $x$ .

This maintains the invariant that at any point through the input the reservoir contains a uniform random sample of the elements processed so far.

from itertools import islice
from random import random
from typing import Iterable

import numpy as np


def method_4(iterable: Iterable, n: int):
    iterable = iter(iterable)
    # Initialise reservoir with first n elements
    reservoir = np.array(list(islice(iterable, n)))
    # Assign random numbers to the first n elements
    u_array = np.array([random() for _ in range(n)])

    # Track the index of the largest u-value
    u_max_idx = u_array.argmax()
    for x in iterable:
        u = random()
        if u < u_array[u_max_idx]:
        # Replace largest u-value
            u_array[u_max_idx] = u
        # Replace corresponding reservoir element
            reservoir[u_max_idx] = x
        # Update largest u-value
            u_max_idx = u_array.argmax()

    return reservoir

As with Algorithm R, Method 4 requires $O (N)$ time and due to the secondary $u$ -value structure may require more memory. Why discuss an arguably worse algorithm? Method 4
forms the basis for another algorithm which is significantly more efficient than both
that have been discussed so far.

Algorithm L (Geometric Jumps)

Algorithm L takes the idea from Method 4 and greatly improves it's runtime to $O (n l o g (n) (1 + l o g (N / n)))$ using several (in my opinion rather genius) mathematical insights from Devroye⁴ and Li⁵. The improvement comes from following the second framework and "jumping" through the input directly to the elements which are to be added to the reservoir.

Insight 1: The size of the "jumps" between reservoir insertions follows a geometric
distribution.

Let $W$ be the largest value in the $u$ -array from Method 4 and $S$ be the number of
elements to be skipped (the "jump" size) before the next reservoir insertion. Given $W$ ,

\lfloor log(random()) / log(1 - W)\rfloor

Insight 2: The $u$ -value of the next record to enter the reservoir has the $U (0, W)$
distribution.

As stated in Method 4, a new element enters the reservoir when it's $u$ -value is less than or equal to $W$ . Insight 2 therefore follows from the fact that $u$ -values are generated randomly with distribution $U (0, 1)$ .

Insight 3: Maintaining the $u$ -array is not necessary - only the variable $W$ is
required.

After the reservoir has been populated with the first $n$ elements in the input, the $u$ -array would contain a sample of $n$ values from $U (0, 1)$ . $W$ , therefore, has the same distribution as the largest value in a sample of size $n$ from $U (0, 1)$ .

Subsequently, the next value, $W^{'}$ , has the same distribution as the largest value in a sample of size $n$ from $U(0, W).

So the distribution of $W$ is known, but how can this be used to generate a concrete value as part of the sampling algorithm? Well, the how is actually rather simple! Li provides the following expression to generate the largest value from a sample of size $n$ from $U (0, b)$ :

ParseError: KaTeX parse error: Undefined control sequence: \

However, the why requires some relatively involved mathematics and I (as Li did in his
paper) won't go into it here⁶. For now, let's assume the Li is
correct and move on to describe Algorithm L:

Populate the reservoir with the first $n$ elements.
Set $\text{exp}(\text{log}(\text{random}())/n)$ .
Set $\lfloor \text{log}(\text{random}()) / \text{log}(1 - W)\rfloor$ .
Until the input is exhausted:
1. Skip over the next $S$ records.
2. Replace a random element of the reservoir with the $S + 1$ th record.
3. Set $W = W * e x p (l o g (r a n d o m ()) / n)$

from itertools import islice
from math import exp, floor, log
from random import random, randrange
from typing import Any, Iterable, Optional

# Sentinel value to mark End Of Iterator
EOI = object


def algorithm_l(iterable: Iterable, n: int):
    iterable = iter(iterable)
    # Initialise reservoir with first n elements
    reservoir = list(islice(iterable, n))

    w = exp(log(random()) / n)

    while True:
        s = floor(log(random()) / log(1 - w))
        next_elem = nth(iterable, s + 1, default=EOI)

        if next_elem is EOI:
            return reservoir

        reservoir[randrange(0, n)] = next_elem
        w *= exp(log(random()) / n)

    return reservoir


def nth(iterable: Iterable, n: int, default: Optional[Any] = None):
    """Returns the ``n``th item in ``iterable`` or a default value.

    Credit: https://docs.python.org/3.7/library/itertools.html#itertools-recipes
    """
    return next(islice(iterable, n, None), default)

As mentioned previously, Algorithm L has $\text{log}(n)(1 + \text{log}(N/n)))$
runtime. I encourage the reader to take a look through the extensive proof of this by Li
in the original paper⁵.

How can a Reservoir Sampling implementation be practically tested for correctness?

Excellent! We know how multiple different Reservoir Sampling algorithms work and have
the mathematical proofs of their correctness. But what about an actual implementation?
How can we tell if the code follows the algorithm correctly and thus the proofs still
hold? All Reservoir Sampling algorithms share two main correctness properties, based on
the output sample:

Size $n$ .
Uniformly distributed across the input.

The first is trivial to test - run the algorithm in question over the input and check
the output size is equal to $n$ . The second property, on the other hand, requires a different approach. As the sampling is probabilistic (through the use of randomisation) simple example-based⁷ tests are insufficient - each test run will produce a different answer. Statistical tests can be used instead, whereby a test asserts against some statistical property of the result as opposed to it's concrete value. The key to statistical testing is choosing which property to use.

One approach that (in my opinion) fits well with Reservoir Sampling involves using the
output to calculate the Maximum Likelihood
Estimator(MLE) for a
parameter of the input distribution. If the estimate matches (within some tolerance)
the real parameter value then we have strong evidence that the algorithm is implemented
correctly.

Generate input data with a known distribution and parameters.
Sample the data using the algorithm under test.
Calculate the MLE of a parameter of the distribution using the output from step 2.
Test the MLE is within some small tolerance $ϵ\epsilon$ of the real value.

The larger the input and sample sizes, the smaller the tolerance can be.

Assuming we have access to a method of generating the input data, the rate parameter
$λ\lambda$ of an Exponential
distribution $Exp(λ)\text{Exp}(\lambda)$
works nicely for Reservoir Sampling as all the information required for the formula to
calculate the MLE for $λ\lambda$ is already available. Given random variable $\text{Exp}(\lambda)$ and $n$ samples⁸ $x1,…,xnx_1, \ldots, x_n$ from $X$ , the MLE for rate is the reciprocal of the sample mean:

λ^=n∑ixi \hat{\lambda} = \frac{n}{\sum_i x_i}

To use this in a statistical test:

Choose a value for $λ\lambda$ .
Generate input data with distribution $Exp(λ)\text{Exp}(\lambda)$ .
Choose a value for $n$ .
Pass the input into a Reservoir Sampling algorithm to generate a sample $x1,…,xnx_1, \ldots, x_n$ .
Calculate $λ^=n∑ixi\hat{\lambda} = \frac{n}{\sum_i x_i}$ .
Assert $λ^=λ±ϵ\hat{\lambda} = \lambda \pm \epsilon$ .

import pytest
from numpy.random import default_rng


def test_sample_distribution():
    # Test parameters - adjust to affect the statistical significance of the test.
    n = 1000  # Sample size
    population_size = n * 1000  # Input size.
    rate = 0.1  # Exponential distribution rate parameter.
    tolerance = 0.1  # How close to the real rate the MLE needs to be.

    # Generate exponentially distributed data with known parameters
    rng = default_rng()
    population = rng.exponential(
        1 / rate,  # numpy uses the scale parameter which is the inverse of the rate
        population_size,
    )
    population.sort()

    # Run Reservoir Sampling to generate a sample
    sample = reservoir_algorithm(population, N)

    # Calculate the MLE for the rate parameter of the population distribution
    rate_mle = len(sample) / sum(sample)

    assert pytest.approx(rate_mle, tolerance) == rate

Conclusion

Reservoir Sampling describes a common framework for probabilistic sampling algorithms
that can be useful in situations where a sample from an input of unknown size is required and where the complete input will not fit into memory. The progression from the simplest implementation of Algorithm R to Algorithm L (and others not discussed here) clearly demonstrates how re-framing a problem can lead to powerful insights that make a real difference. In addition, I think this is a great example where "doing the simple thing" first leads to better results in the future - I very much doubt that Li would ever have come up with Algorithm L had Algorithm R not been proposed first.

~ log(N/n))). ACM Trans. Math. Softw. 20, 4 (Dec. 1994), 485–486.
DOI:https://doi.org/10.1145/198429.198435

Jeffrey S. Vitter. 1985. Random sampling with a reservoir. ACM Trans. Math.
Softw. 11, 1 (March 1985), 37–57. DOI:https://doi.org/10.1145/3147.3165 ↩
Donald E. Knuth. 1997. The art of computer programming, volume 2 (3rd ed.):
Seminumerical Algorithms, 144. Addison-Wesley Longman Publishing Co., Inc., USA. ↩
FAN, C. W., MULLER, M. E., AND REZUCHA, I. Development of sampling plans by
using sequential (item by item) selection techniques and digital computers, j. Am.
Statist. Assoc. 57 (June 1962), 387-402. ↩
Devroye, Non-Uniform Random Variate Generation. New York, NY: Springer New
York, 1986. ↩
Kim-Hung Li. 1994. Reservoir-sampling algorithms of time complexity O(n(1 + ↩
But I may write up a separate article for it! ↩
Given some input $x$ , the algorithm always produces some output
$y$ . ↩
Assuming
IID. ↩

Container with Most Water

Ben Steadman — Thu, 11 Jun 2020 19:38:17 +0000

This post originally appeared on steadbytes.com

A solution and, importantly, a proof for LeetCode Problem 11 - Container with Most Water.

This problem intrigued me as, unlike many "LeetCode-esque" problems, little advanced or specific algorithms theory is required for an optimal $O (n)$ solution. That is, there was no algorithmic or mathematical trick that one needed to solve the problem. Instead, the solution relies on a subtle insight that leads to a relatively simple algorithm. This, in my opinion, is the real skill behind designing algorithms that can be gained from "LeetCode-esque" problems - as opposed to re-writing the same handful of algorithms over and over again.

The algorithm itself is fairly simple to understand intuitively, however understanding why it works is less so. In this article, I present my solution (which is essentially identical to many others you may find) and provide a formal proof of the key insight; something I have been unable to find a satisfactory example of.

I encourage the reader to visit LeetCode and attempt to solve the problem before reading further.

Problem

Given $n$ non-negative integers $a_{1}, a_{2}, ..., a_{n}$ , where each represents a point at coordinate $i, a_{i})$ . $n$ vertical lines are drawn such that the two endpoints of line $i$ is at $i, a_{i})$ and $(i, 0)$ . Find two lines, which together with x-axis forms a container, such that the container contains the most water.

Note: You may not slant the container and $\geq 2$

— https://leetcode.com/problems/container-with-most-water

Example:

Input: [1, 8, 6, 2, 5, 4, 8, 3, 7]
Output: 49

Solution

impl Solution {
    pub fn max_area(height: Vec<i32>) -> i32 {
        let mut l = 0;
        let mut r = height.len() - 1;
        let mut max_area = 0;

        while l < r {
            let width = (r - l) as i32;
            if height[l] < height[r] {
                max_area = max(max_area, height[l] * width);
                l += 1;
            } else {
                max_area = max(max_area, height[r] * width);
                r -= 1;
            }
        }
        max_area
    }
}

Using two pointers, the entire array of heights is processed in a single pass from both ends simultaneously; thus $O (n)$ . On each iteration, the area of the container between l and r is calculated and the lower-heighted of the two is moved towards the other. Either can be moved in the case that i == j (so long as it is consistent) - here I chose to move r. The maximum are

The key idea is that for any two positions of l and r, the corresponding container area is larger than that of any container produced by moving the pointer with the larger height towards the other. All such containers therefore need not be considered.

Proof

Definitions

Let $a_{i}$ be the height of the line at index $i$ and $W (i, j)$ , $H (i, j)$ , $A (i, j)$ be the width, height and area, respectively, formed by the container between $a_{i}$ and $a_{j}$ :

W (i, j) = i - j

H(i, j) = min(a_{i}, a_{j})

\cdot H(i,j)

Loop Invariant

The "key idea" previously mentioned.

Hypothesis: For any $i, j$ where $i < j$ , $A (i, j)$ is greater than the area produced by moving either $j$ towards $i$ if $a_{i} < a_{j}$ or $i$ towards $j$ otherwise:

ai<ajmax({A(i,j),...,A(j−1,j})else\begin{cases}max(\{A(i, j),...,A(i, i+1\}) &\text{if $a_{i} < a_{j}$}\\max(\{A(i, j),...,A(j-1, j\}) &\text{else}\end{cases}

When $a_{i} < a{j}$ :

i′=ii^\prime=i

j′=j−1j^\prime = j - 1

The container width, by definition, has decreased since $j$ has moved closer to $i$ :

j^\prime) < W(i, j)

Since $a_{i} < a_{j}$ , there are two cases for the height of the container $j^\prime)$ :

ai≥aj′⇒H(i,j′)=aj′≤H(i,j)a_{i} \geq a_{j^\prime} \Rightarrow H(i, j^\prime) = a_{j^\prime} \leq H(i, j)

ai<aj′⇒H(i,j′)=ai<H(i,j)a_{i} < a_{j^\prime} \Rightarrow H(i, j^\prime) = a_{i} < H(i, j)

⇒H(i,j′)≤H(i,j)\Rightarrow H(i, j') \leq H(i, j)

Thus,

j^\prime) < A(i, j) \quad \text{by definition of \(A\)}

The same process applies to the case where $a_{i} > a_{j}$ .

Termination

Since the result is returned immediately after exiting the single while loop, proving termination is equivalent to proving the termination of this loop.

The while loop terminates when the expression l < r evaluates to false:

while l < r {
    // ...
}
// l >= r

The loop contains and if-else expression which defined the only two possible code paths:

while l < r {
    let width = (r - l) as i32;
    if height[l] < height[r] {
        max_area = max(max_area, height[l] * width);
        l += 1; // 1.
    } else {
        max_area = max(max_area, height[r] * width);
        r -= 1; // 2.
    }
}

The lines indicated 1. and 2. are the only lines that affect the while loop condition - calculations of width and max_area do not alter the values of l and r. Therefore, there are two relevant cases:

while l < r {
    if some_condition {
        l += 1;
    } else {
        r -= 1;
    }
}

Either, l is incremented or r is decremented. Since l is initialised to 0 and r to the length of the height array the two must converge and thus the while loop must terminate.

More formally, with respect to pointers $l$ and $r$ , each iteration of the while produces new values of $l$ and $r$ as follows:

ai<aj(l,r−1)else(l^\prime, r^\prime) =\begin{cases}(l + 1, r) &\text{if $a_{i} < a_{j}$}\\(l, r - 1) &\text{else}\end{cases}

Since $l$ and $r$ are initialised to $0$ and $n$ , where $n > 0$ , the two values converge and thus loop exit condition $l < r$ is satisfied.

Python, Spark and the JVM: An overview of the PySpark Runtime Architecture

Ben Steadman — Sun, 03 May 2020 13:33:27 +0000

This post originally appeared on steadbytes.com

PySpark seemingly allows Python code to run on Apache Spark - a JVM based computing framework. How is this possible? I recently needed to answer this question and although the PySpark API itself is well documented, there is little in-depth information on its implementation. This article contains my findings from diving into the Spark source code to find out what's really going.

Spark vs PySpark

For the purposes of this article, Spark refers to the Spark JVM implementation as a whole. This includes the Spark Core execution engine as well as the higher level APIs that utilise it; Spark SQL, Spark Streaming etc. PySpark refers to the Python API for Spark.

This distinction is important to understand the motivation for this article:

Spark is a system that runs on the JVM (usually) across multiple machines. PySpark enables direct control of and interaction with this system via Python.

How does this work? Take a look at this visual¹ "TL;DR" and then read on to find out what it all means:

PySpark Execution Model

The high level separation between Python and the JVM is that:

Data processing is handled by Python processes.
Data persistence and transfer is handled by Spark JVM processes.

The Python driver program communicates with a local JVM running Spark via Py4J². Spark workers spawn Python processes, communicating results via TCP sockets.

`SparkContext` and Py4J

A PySpark driver program begins by instantiating a SparkContext; either
directly or indirectly using a SparkSession:



# Direct
from pyspark import SparkContext

sc = SparkContext(
    master="spark-cluster.steadbytes.com", appName="yet_another_word_count"
)

# Indirect - SparkSession manages an underlying SparkContext
from pyspark.sql import SparkSession

spark = (
    SparkSession.builder.master("spark-cluster.steadbytes.com")
    .appName("yet_another_word_count")
    .getOrCreate()
)

In the driver program, pyspark.SparkContext executes spark-submit in a subprocess (yes, that spark-submit) in to initialise a local Spark JVM process:



# ...
script = "./bin/spark-submit.cmd" if on_windows else "./bin/spark-submit"
command = [os.path.join(SPARK_HOME, script)]
# ...
proc = Popen(command, stdin=PIPE, preexec_fn=preexec_func, env=env)
# ...

Before executing spark-submit, a temporary file is created and it's name is exported as an environment variable:



# ...
conn_info_dir = tempfile.mkdtemp()
try:
    fd, conn_info_file = tempfile.mkstemp(dir=conn_info_dir)

    env = dict(os.environ)
    env["_PYSPARK_DRIVER_CONN_INFO_PATH"] = conn_info_file
# ...

Subsequently, spark-submit instantiates a PythonGatewayServer to initialise a Py4J server and write the Py4J server connection details to this file:



// ...
val gatewayServer: Py4JServer = new Py4JServer(sparkConf)
gatewayServer.start()
val boundPort: Int = gatewayServer.getListeningPort
// ... 
val connectionInfoPath = new File(sys.env("_PYSPARK_DRIVER_CONN_INFO_PATH"))
val dos = new DataOutputStream(new FileOutputStream(tmpPath))
dos.writeInt(boundPort)

val secretBytes = gatewayServer.secret.getBytes(UTF_8)
dos.writeInt(secretBytes.length)
dos.write(secretBytes, 0, secretBytes.length)
dos.close()
// ...

The Python driver can then read the contents of the file to establish a Py4J gateway to enable communication between the Python driver and the local Spark JVM process:



# ...
with open(conn_info_file, "rb") as info:
    gateway_port = read_int(info)
    gateway_secret = UTF8Deserializer().loads(info)
# ...

Spark Workers and the `PythonRDD`

A Spark program defines a series of transformations/actions to be performed on some data:



lines = spark.read.text("/path/to/a/big/file.txt").rdd.map(lambda r: r[0])
words = lines.flatMap(lambda line: line.split(' '))
pairs = words.map(lambda word: (word, 1))
word_counts = pairs.reduceByKey(operator.add)

for (word, count) in word_counts.collect():
    print(f"{word}: {count}")

Any functions passed as arguments to Spark RDD operations in the driver program (e.g. map, flatMap, reduceByKey in the above example) are serialised using cloudpickle.

These are then shipped to Spark worker machines by PythonRDD objects which spawn Python processes; serialised code is deserialised, executed and results are sent back to the PythonRDD in the JVM via a TCP socket³.

Summary

There are (as always) further details and intricacies involved in the execution of a PySpark program, however the above should be sufficient to at least understand where the code is running and how different components communicate.

I make no claims to being an artist; this is reaching the limit of my artistic skill. ↩
In depth discussion of how Py4J works is out of scope for this article; please refer to the documentation for further details. ↩
The implementation of the socket server itself is out of scope for this article, however I encourage the reader to take a look at the source if interested. ↩

TPP Topic 38: Programming by Coincidence

Ben Steadman — Sat, 16 Nov 2019 12:20:12 +0000

This post originally appeared on steadbytes.com

See the first post in The Pragmatic Programmer 20th Anniversary Edition series for an introduction.

Exercise 25

A data feed from a vendor gives you an array of tuples representing key-value pairs. The key of DepositAccount will hold a string of the account number in the corresponding value:

[
 ...
 {:DepositAccount, "564-904-143-00"}
 ...
]

It worked perfectly in test on the 4-core developer laptops and on the 12-core build machine, but on the production servers running in containers, you keep getting the wrong account numbers. What’s going on?

Assuming that the key-value pairs are read into a hash map (or similar) data structure, for example in Clojure:

user=> (into {} [[:DepositAccount "564-904-143-00"] [:OtherKey "Other Value"]])
{:DepositAccount "564-904-143-00", :OtherKey "Other Value"}

Then it is possible that the :DepositAccount is being overwritten due to multiple occurrences of a :DepositAccount tuple in the input array. An array of tuples will not enforce uniqueness of the keys, whereas a hash map does. Most hash map implementations will update the value of an existing key when a duplicate insert is performed:

user=> (into {:DepositAccount "111-111-111-11"} [[:DepositAccount "564-904-143-00"]])
{:DepositAccount "564-904-143-00"}

The fact that it worked on laptops and the build machine is a coincidence that (I think) the authors included to lure the reader into a trap!

Exercise 26

You’re coding an autodialer for voice alerts, and have to manage a database of contact information. The ITU specifies that phone numbers should be no longer than 15 digits, so you store the contact’s phone number in a numeric field guaranteed to hold at least 15 digits. You’ve tested in thoroughly throughout North America and everything seems fine, but suddenly you’re getting a rash of complaints from other parts of the world. Why?

Again, the authors are throwing in additional information to demonstrate the trap of coincidence. The ITU's 15 digit length limit is not the cause of the problem, it is the use of a numeric field to store phone numbers in the database. The ITU E.164 standard specifies that the international call format start with a + sign. Characters such as * or leading zeros are also used in some areas. None of these can be stored in a numeric field.

Exercise 27

You have written an app that scales up common recipes for a cruise ship dining room that seats 5,000. But you’re getting complaints that the conversions aren’t precise. You check, and the code uses the conversion formula of 16 cups to a gallon. That’s right, isn’t it?

The "cup" measurement has a surprising number of interpretations therefore the conversion stated in the exercise is dependent on both location and custom. From Wikipedia:

United States
- Legal cup = 240 ml
- Customary cup = 1/2 of a liquid pint ~= 237 ml
Commonwealth of Nations
- Metric cup = 250 ml
- Canadian cup = 8 imperial fluid ounces ~= 227 ml
- United Kingdom = 10 imperial fluid ounces ~= 284 ml
International
- Latin American cup = 200 ml or 250 ml or either US cups
- Japanese cup = 200 ml (legally) or ~180mL (traditionally)

In this case, the programmer made the assumption that the app will only be used within the United States, converting between customary cups and liquid gallons.

TPP Topic 36: Blackboards

Ben Steadman — Sun, 10 Nov 2019 14:45:34 +0000

This post originally appeared on steadbytes.com

See the first post in The Pragmatic Programmer 20th Anniversary Edition series for an introduction.

Exercise 24

Would a blackboard-style system be appropriate for the following applications? Why, or why not?

Image processing. You’d like to have a number of parallel processes grab chunks of an image, process them, and put the completed chunk back.

Group calendaring. You’ve got people scattered across the globe, in different time zones, and speaking different languages, trying to schedule a meeting.

Network monitoring tool. The system gathers performance statistics and collects trouble reports, which agents use to look for trouble in the system

No. Processing chunks of an image in parallel doesn't have complex interactions and/or feedback between processes; an image produces a fixed set of tasks, where the completion of one task doesn't inform the completion of another task. A simple queue would suffice in this case.
Yes. There are multiple people (processes) operating at different times without direct interaction. However the decisions made by each person feedback into further decisions made by others. For example, one person can put up a time where they're definitely not available which may inform which times other people say they are available.
Yes. The performance statistics and problem reports are gathered both independently and non-deterministically (things happen when they happen). Furthermore, a single statistic or problem report may not be an issue but some combination of statistics of given value and problem reports may signal trouble. Many such combinations are possible and separate agents (human or machine) can search the blackboard for a specific issue.

TPP Topic topic-30: Transforming Programming

Ben Steadman — Sun, 03 Nov 2019 12:47:30 +0000

This post originally appeared on steadbytes.com

See the first post in The Pragmatic Programmer 20th Anniversary Edition series for an introduction.

Exercise 21

Can you express the following requirements as a top-level transformation? That is, for each, identify the input and the output.

Shipping and sales tax are added to an order

Your application loads configuration information from a named file

Someone logs in to a web application

initial_order -> final_order
- initial_order represents the unprocessed data of an order - e.g. the cost and quantity of each item.
- final_order represents an order that has been processed/transformed into a shippable state with shipping and sales tax calculated and included appropriately.
config_file_name -> config
- The configuration file is read and it's contents are transformed into a data structure to be used for configuring the application.
user_credentials -> user_session
- The user provides their login credentials (username/password, OAuth token e.t.c) which are processed to grant access to the web application.

Exercise 22

You’ve identified the need to validate and convert an input field from a string into an integer between 18 and 150. The overall transformation is described by

field contents as string
   → [validate & convert]
       → {:ok, value} | {:error, reason}

Write the individual transformations that make up validate & convert

field contents as string
    -> [convert to integer]
    -> [18 <= value <= 150]
        -> {:ok, value} | {:error, "value not 18 <= value <= 150"}

Exercise 23

In Language X Doesn't Have Pipelines, on page 153 we wrote:

const content = File.read(file_name);
const lines = find_matching_lines(content, pattern);
const result = truncate_lines(lines);

Many people write OO code by chaining together method calls, and might be
tempted to write this as something like:

const result = content_of(file_name)
  .find_matching_lines(pattern)
  .truncate_lines();

What’s the difference between these two pieces of code? Which do you think
we prefer?

The second piece of code uses a fluent interface, whereas the first 'chains' isolated functions together by explicitly passing the result one function call to the next. The difference here is that each function call in the fluent interface is a method on the object returned by the previous call; the object returned by find_matching_lines must implement a truncate_lines method. This introduces tight coupling to the 'pipeline' code and the objects used within it. Changing the pipeline behaviour is more difficult as one needs to decide which object a given transformation should be attached to. For example, say the requirements change such that the output must all be lower case. In the first piece of code, this is trivial:

const content = File.read(file_name);
const lines = find_matching_lines(content, pattern);
const lower_case_lines = to_lower_case(lines);
const result = truncate_lines(lower_case_lines);

In the second piece of code however a decision needs to be made as to where the to_lower_case functionality should be implemented: perhaps it should be on the object returned by File.read, or maybe on the object returned by find_matching_lines. Depending on what those objects actually represent it may not make semantic sense to have a to_lower_case method in the general case. Furthermore, there is the possibility of breaking other code which uses the changed object.

I'm 100% certain that the authors prefer the first piece of code 😄

TPP Topic 29: Juggling the Real World

Ben Steadman — Wed, 23 Oct 2019 17:14:30 +0000

This post originally appeared on steadbytes.com

See the first post in The Pragmatic Programmer 20th Anniversary Edition series for an introduction.

Exercise 19

In the FSM section we mentioned that you could move the generic state machine implementation into its own class. That class would probably be initialized by passing in a table of transitions and an initial state.

Try implementing the string extractor that way.

The generic part of the event/strings_fsm.rb example in the book is the logic to determine the appropriate state transition given an occurrence of some event . In this case the event is a character read from standard input; each new character determined the next state. Actually performing some action given a state transition is not generic and would be implemented by the caller of the generic state machine class.

# fsm.py

from typing import Dict, Hashable, Union, List, Tuple, Iterable

State = Hashable
Event = Hashable


class StateMachine:
    def __init__(
        self,
        transitions: Dict[State, Dict[Union[Event, State], List[State]]],
        default_state: State,
        initial_state: State,
    ):
        self.transitions = transitions
        self.default_state = default_state
        self.initial_state = initial_state

    def process(self, events: Iterable[Event]) -> Tuple[State, State, Event]:
        state = self.initial_state
        for event in events:
            state, action = self.transitions[state].get(
                event, self.transitions[state][self.default_state]
            )
            yield state, action, event

StateMachine.process takes the stream of events as input and returns a generator which, when iterated, yields the next state, action and the current event. The strings finite state machine could be implemented using this as follows:

# strings_fsm.py

from enum import Enum, auto

from fsm import StateMachine


class State(Enum):
    DEFAULT = auto()
    IGNORE = auto()
    LOOK_FOR_STRING = auto()
    START_NEW_STRING = auto()
    IN_STRING = auto()
    COPY_NEXT_CHAR = auto()
    ADD_CURRENT_TO_STRING = auto()
    FINISH_CURRENT_STRING = auto()


TRANSITIONS = {
    State.LOOK_FOR_STRING: {
        '"': [State.IN_STRING, State.START_NEW_STRING],
        State.DEFAULT: [State.LOOK_FOR_STRING, State.IGNORE],
    },
    State.IN_STRING: {
        '"': [State.LOOK_FOR_STRING, State.FINISH_CURRENT_STRING],
        "\\": [State.COPY_NEXT_CHAR, State.ADD_CURRENT_TO_STRING],
        State.DEFAULT: [State.IN_STRING, State.ADD_CURRENT_TO_STRING],
    },
    State.COPY_NEXT_CHAR: {
        State.DEFAULT: [State.IN_STRING, State.ADD_CURRENT_TO_STRING]
    },
}


def find_strings(char_stream):
    fsm = StateMachine(TRANSITIONS, State.DEFAULT, State.LOOK_FOR_STRING)

    for state, action, ch in fsm.process(char_stream):
        if action == State.IGNORE:
            continue
        elif action == State.START_NEW_STRING:
            result = []
        elif action == State.ADD_CURRENT_TO_STRING:
            result.append(ch)
        elif action == State.FINISH_CURRENT_STRING:
            yield "".join(result)

Usage:

>>> for s in find_strings('"foo" bar "baz bat"'):
...     print(s)
foo
baz bat

This version is further abstracted by separating the state machine input from it's processing. The original version read characters directly from standard input within the state machine loop whereas this implementation receives a char_stream parameter intended to be an iterable of characters (strings of length 1 in Python). The state transitions are the same as those given events/strings_fsm.rb in the book except defined using an Enum instead of strings.

The complete implementation including tests and reading from standard input can
be found on GitHub here.

Exercise 20

Which of these technologies (perhaps in combination) would be a good fit for the following situations:

If you receive three network interface down events within five minutes, notify the operations staff.

If it is after sunset, and there is motion detected at the bottom of the stairs followed by motion detected at the top of the stairs, turn on the upstairs lights.

You want to notify various reporting systems that an order was completed.

In order to determine whether a customer qualifies for a car loan, the application needs to send requests to three backend services and wait for the responses.

Event streams
- The consumer of the network interface down events could group the stream of events into 3-tuples of consecutive events. The timestamp of the first and last events in each group would then be compared to determine if the operations staff should be notified.
Pubsub + state machine
- Each motion detector would publish to a channel when it detects motion. A state machine can then subscribe to these channels to determine whether the downstairs motion event was produced prior to the upstairs motion event (with some appropriately sized delay between the two events). The lights would be subscribed to a control channel which the state machine can publish to in order to turn on the lights. The state machine will simply ignore the motion detector events if it is before sunset.
Pubsub
- Each reporting system can subscribe to a completed orders channel. When the order is completed an event is published to the channel and the reporting systems can report to their heart's content.
Event streams
- Each backend service call would be an event in an asynchronous stream - similar to event/rx1/index.js example in the book for fetching users from a REST API. This allows the requests to performed concurrently and therefore improve throughput.

TPP Topic 26: How to Balance Resources

Ben Steadman — Sun, 13 Oct 2019 13:34:41 +0000

This post originally appeared on steadbytes.com

See the first post in The Pragmatic Programmer 20th Anniversary Edition series for an introduction.

Challenge 1

Although there are no guaranteed ways of ensuring that you always free resources, certain design techniques, when applied consistently, will help. In the text we discussed how establishing a semantic invariant for major data structures could direct memory deallocation decisions. Consider how Topic 23, Design by Contract, on page 104, could help refine this idea.

Design by Contract could be used to enforce the semantic invariants for a given data structure. The code used to create, operate on and destroy the data structure can utilise pre and post conditions to ensure that the design techniques for deallocating memory are used. The text suggests that for procedural languages such as C:

... write a module for each major structure that provides standard allocation and deallocation facilities for that structure.

In Chapter 3 of "Writing Solid Code", author Steve Maguire takes this idea further; implementing memory allocation/deallocation bookkeeping in order to codify and enforce correct and safer resource management. Furthermore, throughout the book Maguire utilises design by contract through use of C's assert macro. Using design by contract and the concepts from Maguire's book, I implemented a queue data structure module in C. The full implementation can be found on my GitHub: C Data Structure Memory Management Example.

This approach is certainly not perfect, it is certainly an improvement over uncontrolled usage of malloc and free.

Exercise 17

Some C and C++ developers make a point of setting a pointer to NULL after they deallocate the memory it references. Why is this a good idea?

This is one method to help prevent use after free errors. In C/C++, there is no way of determining whether a pointer is valid - i.e. it could be pointing to unallocated memory. Using an invalid pointer can cause many issues, most notably corrupting valid data and potential execution of arbitrary code. Setting a pointer to NULL will reliably generate runtime error if it is dereferenced - crashing the program instead of allowing use after free errors to go unnoticed.

Exercise 18

Some Java developers make a point of setting an object variable to NULL after they have finished using the object. Why is this a good idea?

Java uses garbage collection to automatically manage deallocation of memory without the programmer explicitly doing so as in C. Part of this garbage collection is reference counting, whereby an object will be deallocated if there are no references to it in the current state of the program. Setting an object variable to NULL decrements the reference count for that particular object so that it can be garbage collected. In long running applications, such as a web server, this can be important to prevent memory utilisation from constantly increasing over time due unused objects continuing to be referenced within the program.

Algorithm Optimisation: A Real Example

Ben Steadman — Sat, 28 Sep 2019 13:26:04 +0000

This post originally appeared on steadbytes.com

In an interview, I was asked to analyse the complexity (and thus relative performance)
of a piece of C code extracted from functioning production software. As a follow up, I will demonstrate the process of identifying a performance problem and implementing a solution to it.

The code had originally been given as part of a pre-interview exam, where general issues with the code were to be discussed. I won't be posting solutions to the pre-interview exam question (there will be potential problems with the code) and I have changed the structure of the code so it's not searchable for future candidates. However, the core algorithm represents an interesting 'real world' optimisation problem to focus on in this post.

Original `read_file` Algorithm

As the name suggests, read_file reads a file from disk into memory; storing the data in
an array and returning the total number of bytes read:

#include <unistd.h>
#include <fcntl.h>
#include <string.h>
#include <stdlib.h>

#define BUFSIZE 4096

size_t read_file(int fd, char **data)
{
    char buf[BUFSIZE];
    size_t len = 0;
    ssize_t read_len;
    char *tmp;

    /* chunked read from fd into data */
    while ((read_len = read(fd, buf, BUFSIZE)) > 0)
    {
        tmp = malloc(len + read_len);
        memcpy(tmp, *data, len);
        memcpy(tmp + len, buf, read_len);
        free(*data);
        *data = tmp;
        len += read_len;
    }
    return len;
}

Algorithmic Complexity

read_file has O(n²) space and time complexity caused by the memcpy
operations within the while loop:

while ((read_len = read(fd, buf, BUFSIZE)) > 0)
{
    /* allocate space for both existing and new data */
    tmp = malloc(len + read_len);
    /* copy existing data */
    memcpy(tmp, *data, len);
    /* copy new data */
    memcpy(tmp + len, buf, read_len);
    free(*data);
    *data = tmp;
}

All previously read data is copied into a new array in each iteration:

+-+
| |
+-+
 | copy existing data
 v
+-+   +-+
| |   | |
+-+   +-+
 |     |
 v     | read new data
+--+   |
|  | <-+
+--+
 |
 v
+--+   +-+
|  |   | |
+--+   +-+
  |     |
  v     |
+---+   |
|   | <-+
+---+
 |
 v
+---+    +-+
|   |    | |
+---+    +-+
  |       |
  v       |
+----+    |
|    | <--+
+----+

...

Identifying a Performance Problem

Finding the complexity of an algorithm is great to get a relative idea of it's performance will scale with input size. However this is not the same as real world run time; an O(n²) algorithm may not be an issue if it only ever receives small input, for example. Before beginning optimisation the code under question should always be benchmarked and/or profiled. This helps to identify if a problem actually exists and whether a proposed solution is effective.

"The real problem is that programmers have spent far too much time worrying about efficiency in the wrong places and at the wrong times; premature optimization is the root of all evil (or at least most of it) in programming"

-- Donald Knuth, The Art of Computer Programming

The main entry point to the program simply uses read_file to read from standard input and prints the total number of bytes read:

#include <stdio.h>
#include <unistd.h>
#include <fcntl.h>
#include <string.h>
#include <stdlib.h>

#define BUFSIZE 4096

size_t read_file(int fd, char **data);

int main(int argc, char *argv[])
{
    char *data = NULL;

    size_t len = read_file_improved(STDIN_FILENO, &data);

    printf("%ld\n", len);

    return 0;
}

Running the program with randomly generated text files of increasing size clearly demonstrates the O(n²) complexity:

An input size of 0.5Gb takes over an hour to complete!

Improved Algorithm

The issue with read_file is the repeated copying of data - copying each byte only
once would give O(n) complexity. Using a linked list, pointers to each chunk read from the file can be stored in order; creating a new node in each iteration of the loop. After the loop terminates, the linked list can be traversed once and all chunks copied in the final data array. In pseudocode:

typedef struct node_t
{
    char *data; /* chunk of data read from file */
    int len; /* number of bytes in chunk */
    struct node_t *next; /* pointer to next node */
} node_t;

size_t read_file_improved(int fd, char **data)
{
    char buf[BUFSIZE];
    ssize_t read_len;
    size_t len = 0;

    /* build linked list containing chunks of data read from fd */
    node_t *head = NULL;
    node_t *prev = NULL;
    while ((read_len = read(fd, buf, BUFSIZE)) > 0)
    {
        node_t *node = malloc(sizeof(node_t));
        node->data = malloc(read_len);
        memcpy(node->data, buf, read_len);
        node->len = read_len;
        node->next = NULL;

        /* first chunk */
        if (head == NULL)
        {
            head = node;
        }
        /* second chunk onwards */
        if (prev != NULL)
        {
            prev->next = node;
        }
        prev = node;
        len += read_len;
    }

    /* copy each chunk into data once only */
    *data = malloc(len);
    char *p = *data;
    node_t *cur = head;
    while (cur != NULL)
    {
        memcpy(p, cur->data, cur->len);
        p += cur->len;
        cur = cur->next;
    }

    return len;
}

Algorithmic Complexity

Building the linked list (assuming constant overhead for node_t structs) is O(n); each byte from the file is read from the file into an array and then never touched again:

    file reads
 +------+------+
 |      |      |
 V      V      V
+-+    +-+    +-+
| | -> | | -> | | ...
+-+    +-+    +-+

Arrows between boxes represent linked list node pointers

The final step to copy chunks from the linked list into data is also O(n); traversing pointers in the linked list is O(1) and each byte in each chunk is copied once.

Together, that makes O(2n) ~= O(n)

Performance Comparison

To benchmark each implementation, the main entry point is adapted to take an optional --slow argument to toggle between using the improved algorithm by default and the original:

int main(int argc, char *argv[])
{
    size_t len;
    char *data = NULL;
    /* default to improved implementation*/
    if (argc == 1)
    {
        len = read_file_improved(STDIN_FILENO, &data);
    }
    /* optionally specify original implementation for comparison purposes */
    else if (argc == 2 && !strcmp(argv[1], "--slow"))
    {
        len = read_file(STDIN_FILENO, &data);
    }
    else
    {
        fprintf(stderr, "usage: read_file [--slow]\nReads from standard input.");
        return 1;
    }
    /* do some work with data */
    printf("%ld\n", len);

    return 0;
}

Running the same benchmark of randomly generated text files produces
quite clear results:

Conclusion

Understanding algorithmic complexity is key to implementing well performing programs. However, it should also be coupled with performance benchmarks to avoid premature optimisation. In this post I have demonstrated how to:

Identify a performance problem
Determine the complexity of an algorithm
Re-design an algorithm to be more performant
Prove the validity of an optimisation

Full code including benchmark script and analysis can be found on GitHub here.

TPP Topic 25: Assertive Programming

Ben Steadman — Sat, 21 Sep 2019 14:44:17 +0000

This post originally appeared on steadbytes.com

See the first post in The Pragmatic Programmer 20th Anniversary Edition series for an introduction.

Exercise 16

A quick reality check. Which of these “impossible” things can happen?

A month with fewer than 28 days

Error code from a system call: can’t access the current directory

In C++: a = 2; b = 3; but (a + b) does not equal 5

A triangle with an interior angle sum ≠ 180°

A minute that doesn’t have 60 seconds

(a + 1) <= a

All of them!

In 1752, Great Britain changed from using the Julian calendar to the Gregorian calendar. This required a correction of 11 days; causing September to have only 19 days.
There are no certainties when it comes to file systems (take a look at Dan Luu's excellent talk Files are fraught with peril). To name but a few reasons for this particular error: the directory could have been deleted by another process, the process making the system call may not have permissions to access the directory or the drive where the directory is located could have been disconnected.
Without extra care, concurrency could cause a to be updated before (a + b) is executed.
In curved geometries the interior angles of a triangle do not sum to 180°. In elliptic geometry the sum will be greater than 180° and in hyperbolic the sum will be less than 180°.
Dates and times are also fraught with peril! In this case, leap seconds can cause minutes with 59 or 61 seconds.
- Note: at the time of writing, all leap seconds have been positive - causing a minute to have 61 seconds. A negative leap second would occur if the earth's rotation were to increase in speed instead of decrease as it has been.
Integer overflow could cause the value of a + b to be negative. The exact scenarios under which overflow can occur depend on the language and the compiler/interpreter being used; the Integer overflow page on Rosetta Code has examples from many different languages. For this specific case I present an example using Go (GitHub):

// overflow.go

package main

import "fmt"

func main() {
    var a int32 = 2147483647
    fmt.Printf("a = %d\n", a)
    fmt.Printf("(a + 1) <= a = false, got %t\n", (a+1) <= a)
    fmt.Printf("(%d + 1) = 21474836478, got %d\n", a, a+1)
}

Output:

$ go run overflow.go
a = 2147483647
(a + 1) <= false, got true
(2147483647 + 1) = 21474836478, got -2147483648

Program exited.

TPP Topic 23: Design by Contract

Ben Steadman — Thu, 19 Sep 2019 10:58:42 +0000

This post originally appeared on steadbytes.com

See the first post in The Pragmatic Programmer 20th Anniversary Edition series for an introduction.

Challenge 1

Points to ponder: If DBC is so powerful, why isn’t it used more widely? Is it hard to come up with the contract? Does it make you think about issues you’d rather ignore for now? Does it force you to THINK!? Clearly, this is a dangerous tool!

Design by contract leads to similar arguments as the great dynamic vs static typing debate:

Code conciseness
- Dynamic languages can be more concise
Speed of iteration
- Dynamic languages is often quicker (though there is a limit)
- Static languages require more up front consideration
Ease of creating generic code
- Often easier in dynamic languages

I emphasise that none of the above points are absolute truths; there are certainly exceptions to both sides of each and nor are they necessarily my views.

For DBC to be effective, the constraints of a system need to be known and considered. Often, one or both of these is not exercised during development. Furthermore, I think there is a notion that (similar to static types) DBC 'cements' the implementation of code and thus prevents agile, iterative development. If DBC is implemented carefully ,however, this is not the case. The implementation should be decoupled from the constraints imposed by the contracts and changing contracts as the requirements of
software changes should be a normal practice. Achieving this however does require more up front work.

Exercise 14

Design an interface to a kitchen blender. It will eventually be a web-based, IoT-enabled blender, but for now we just need the interface to control it. It has ten speed settings (0 means off). You can’t operate it empty, and you can change the speed only one unit at a time (that is, from 0 to 1, and from 1 to 2, not from 0 to 2).

Here are the methods. Add appropriate pre- and postconditions and an invariant.

int getSpeed()
void setSpeed(int x)
boolean isFull()
void fill()
void empty()

See dbc_blender for full code and tests.

I've chosen to implement this in Clojure due to it's built in support for pre- and postconditions provided by condition maps and the excellent spec which is ideal for imposing constraints on data (also because I like Clojure).

To better fit the functional approach of Clojure, I have made a few changes to the proposed OOP style interface:

The blender is modelled as a map instead of a class

  => {::speed 5 ::full true}

getSpeed and isFull are not implemented as the keys of the blender map can be used directly

  ; getSpeed
  => (::speed {::speed 5 ::full true})
  5
  ; isFull
  => (::full {::speed 5 ::full true})
  true

setSpeed, fill and empty functions are not void; instead returning a full blender map updated to reflect the desired changes in order to avoid mutation of state

Before diving into any implementations, here's the namespace definition that any code
examples will be working within:

(ns dbc-blender.core
  (:gen-class)
  (:require [clojure.spec.alpha :as s]
            [clojure.math.numeric-tower :as math]))

First off, a valid range for the blender speed needs to be determined. The question states that
speed must increment one unit at a time; suggesting that an integer data type makes sense (though
further enforcement is required). Furthermore, a speed cannot be negative and (at least on all blenders I've ever seen) there is a sensible limit to the speed value.

; turn it up to 11!
(def max-speed 11)

; speed is an integer and in correct range
(s/def ::speed (s/and int? #(and (>= % 0) (<= % max-speed))))

Next, the blender map has a few constraints:

Requires two keys ::speed and ::full
::full must be of boolean type
The blender should only be on when full. This is not explicitly stated in the exercise, but it is a sensible design constraint that I've chosen to apply

(s/def ::full boolean?)
(s/def ::blender (s/and (s/keys :req [::speed ::full])
                        #(if (> (::speed %) 0) ; should only be on when full
                           (::full %)
                           true)))

set-speed constraints:

Take as input a valid ::blender and a valid ::speed
Return a new valid ::blender with an updated ::speed value
::speed can only be increased by one unit

(defn set-speed [blender x]
  {:pre [(s/valid? ::blender blender)
         (s/valid? ::speed x)
         (::full blender)
         (= (math/abs (- (::speed blender) x)) 1)] ; increase in single increments
   :post [(= (::speed %) x) ; speed was set
          (s/valid? ::blender %)]}
  (assoc blender ::speed x))

fill constraints:

Take as input a valid ::blender
Return a new valid ::blender that is full
Blender cannot already be full
Blender cannot be switched on
- Potentially dangerous
- Make a mess

(defn fill [blender]
  {:pre [(s/valid? ::blender blender)
         (= (::speed blender) 0) ; can't fill a spinning blender
         (not (::full blender))] ; can't fill an already full blender
   :post [(s/valid? ::blender %)
          (::full %)]} ; blender was filled
  (assoc blender ::full true))

empty constraints:

Take as input a valid ::blender
Return a new valid ::blender that is empty
Blender cannot already be empty
Blender cannot be switched on (same reasoning as for fill)

(defn empty [blender]
  {:pre [(s/valid? ::blender blender)
         (= (::speed blender) 0) ; can't empty a spinning blender
         (::full blender)] ; can't empty an empty blender
   :post [(s/valid? ::blender %)
          (not (::full %))]} ; blender was emptied
  (assoc blender ::full false))

Take a look at the tests for comprehensive examples of correct and incorrect usage
of the interface and how they are enforced by the specified contracts.

Exercise 15

How many numbers are in the series 0, 5, 10, 15, …, 100?

I was slightly confused by this question at first; what has a simple series got
to do with DBC? However, it's purpose is to challenge the reader to consider the precise
specification of the series. From the solution provided by the authors:

If you said 20, you just experienced a fence post error (not knowing whether to
count the fenceposts or the spaces in between them)

An answer of 20 implies a series exclusive of either the start or end values:

0, 5, 10, 15, ..., 95
5, 10, 15, ..., 100

Here's some Python to demonstrate each of these, remember that the range function
is start inclusive and end exclusive:

>>> len(range(0, 105, 5)) # correct
21

>>> len(range(0, 100, 5)) # incorrect: excludes the final element 100
20

>>> len(range(5, 105, 5)) # incorrect: excludes the starting element 0
20

TPP Topic 21: Text Manipulation

Ben Steadman — Mon, 16 Sep 2019 18:08:52 +0000

This post originally appeared on steadbytes.com

See the first post in The Pragmatic Programmer 20th Anniversary Edition series for an introduction.

Exercise 11

You’re rewriting an application that used to use YAML as a configuration language. Your company has now standardized on JSON, so you have a bunch of .yaml files that need to be turned into .json. Write a script that takes a directory and converts each .yaml file into a corresponding .json file (so database.yaml becomes database.json, and the contents are valid JSON).

Conversion between YAML and JSON can be done easily in Python using PyYAML and the standard library json module. PyYaml and json both convert basic Python objects (dict, list, str, int e.t.c) to and from YAML and JSON respectively, both providing an almost identical API.

Algorithm outline:

Find all YAML files in a given directory
Load a YAML file from disk into a Python object using PyYaml
Serialize the Python object to JSON
Write the serialized JSON to a new .json file
Delete the YAML file
Repeat steps 2-5 for all YAML files from step 1

import json
from pathlib import Path

import yaml

def main(d: Path):
    # make sure we have a valid directory to search in
    assert d.exists()
    assert d.is_dir()

    for yaml_f in d.glob("*.yaml"):
        # load original YAML data
        with yaml_f.open() as f:
            data = yaml.safe_load(f)
        # write data to new json file
        json_f = d / f"{yaml_f.stem}.json"
        with json_f.open(mode="w") as f:
            json.dump(data, f, indent=4)
        # delete original YAML file
        yaml_f.unlink()
        print(f"{yaml_f} -> {json_f}")

See yaml_to_json.py on GitHub for the full CLI script.

Exercise 12

Your team initially chose to use camelCase names for variables, but then changed their collective mind and switched to snake_case. Write a script that scans all the source files for camelCase names and reports on them.

In a real world scenario where I just need to accomplish the task, I would definitely solve this with egrep:

$ egrep -rn --color "\b[a-z]+((\d)|([A-Z0-9][a-z0-9]+))+([A-Z])?" /path/to/source/directory

However, for educational purposes, I have implemented this functionality in a Python script along with unit tests. See camel_finder.py on GitHub for the full CLI script and tests.

The task breaks down into the following high level steps:

Iterate over a sequence of files (provided as arguments to the script)
Iterate over the lines of each file
Identify any camelCase strings in each line
Output a report of each identified camelCase string

Steps 1 and 2 are trivial in Python:

import sys
from pathlib import Path


def main(files: List[Path]):
    for file in files:
        with file.open() as f:
            for line in f:
                # do steps 3 and 4


if __name__ == "__main__":
    if len(sys.argv) == 1:
        err_exit("Usage: ./camel_finder.py FILES...")

    main([Path(f) for f in sys.argv[1:]]

Step 3 can be achieved with a regular expression as shown in the egrep example. For
this exercise I am using the Google Java style guide definition of camelCase for lower camelCase - not PascalCase.

import re


CAMEL_RE = re.compile(
    (
        # 1st character must be lower case
        r"\b[a-z]+"
        # followed by a single digit
        # OR upper case character/number followed by lower case characters or number
        r"((\d)|([A-Z0-9][a-z0-9]+))"
        # final character *may* be upper case
        r"+([A-Z])?"
    )
)

Demo of this regex: https://regex101.com/r/paY5R6/1/

Using CAMEL_RE, the positions of camelCase substrings can be extracted from a string. The representation of a given line and the positions of any camelCase
substrings contained within it are provided by two NamedTuple objects: MatchGroup
and Match. These are inspired by the standard library re.Match objects which are used to provide the data for MatchGroup
and Match. This provides a separation boundary between the concept of "a line
of text with camelCase substrings" and "a regular expression that matches camelCase
substrings". Thus allowing the underlying mechanism for finding the camelCase substrings
to more easily be changed if necessary.

from typing import Generator, Iterable, NamedTuple


class MatchGroup(NamedTuple):
    start: int
    end: int


class Match(NamedTuple):
    lineno: int
    groups: List[MatchGroup]
    line: str


def find_camel(lines: Iterable[str]) -> Generator[Match, None, None]:
    for i, line in enumerate(lines):
        groups = [MatchGroup(*m.span()) for m in CAMEL_RE.finditer(line)]
        if groups:
            yield Match(i, groups, line)

Example usage:

>>> lines = ["a camelCase line", "thisIs a line with two camelCase words", "a line without camel case"]
>>> for match in find_camel(lines):
...     print(match)
Match(lineno=0, groups=[MatchGroup(start=2, end=11)], line='a camelCase line')
Match(lineno=1, groups=[MatchGroup(start=0, end=6), MatchGroup(start=23, end=32)], line='thisIs a line with two camelCase words')

Step 4 (without installing third party packages) involves using ANSI colour escape codes. To emulate the --color option of grep, each matched
line should be output with the following format:

<optional purple filename>:<green line number>: <white text> <red camelCase match> <white text>...

For example:

<span>/path/to/my/file.txt</span>:<span>5</span>: this line has <span>camelCase</span> words as <span>subStrings</span>.

The 'pretty match' string is built up by iterating through each MatchGroup of a
given Match and extracting a slice of the original line containing non-matched text
before the position of the MatchGroup, extracting the slice of the original line
where at the position of the MatchGroup and adding the red escape code : \033[31m.

Note that after each use of a colour escape code, the colour is reset using \033[m

def pretty_match(m: Match, filename: str = None) -> str:
    """
    Build a 'pretty' string representation of `m`, with coloured text and line
    numbers; optionally prefixed with `filename`.

    Colours:
        - Line numbers = green
        - Matches = red
        - Filenames = purple
        - Non-matching text = white
    """
    pretty_name = f"\033[35m{filename}\033[m:" if filename else ""
    l = []
    prev = 0
    for g in m.groups:
        # text up until match in white, match in red
        l.append((f"{m.line[prev:g.start]}"f"\033[31m{m.line[g.start:g.end]}\033[m"))
        prev = g.end
    l.append(m.line[prev:])
    return "".join([f"{pretty_name}\033[32m{m.lineno}\033[m:"] + l)

Bringing it all together in the original main function:

def main(files: List[Path]):
    """
    Print a report on the locations of all camelCase strings in `file`. See
    `pretty_match` for output format.
    """
    show_filenames = len(files) > 1
    for file in files:
        with file.open() as f:
            for m in find_camel(f):
                print(pretty_match(m, filename=file if show_filenames else None))

Exercise 13

Following on from the previous exercise, add the ability to change those variable names automatically in one or more files. Remember to keep a backup of the originals in case something goes horribly, horribly wrong.

Again, see camel_finder.py on GitHub for the full CLI script and tests.

Once a camelCase substring has been found, converting it to snake_case requires two steps:

Insert an _ character between each camelCase 'hump'
- camelCase -> camel_Case
- camelCamelCase -> camel_Camel_Case
Convert to lower case

Regex substitution using a capture group can be used for step 1 by matching and
capturing a 'hump' character:

camel(C)ase
camel(C)amel(C)ase

Then inserting an _ character before the 'hump' character by referencing the capture group in the substitution: '_\1'.


CONVERT_CAMEL_RE = re.compile(
    (
        # match a nomral 'hump' i.e. camel(C)ase or camel1(C)ase
        r"((?<=[a-z0-9])[A-Z]"
        # match mid-string uppercase humps, ignoring existing underscores
        # i.e. CAMEL(C)ase or HTTP(E)rror
        r"|(?!^)(?<!_)[A-Z](?=[a-z]))"
    )
)


def convert_camel_word(w: str) -> str:
    return CONVERT_CAMEL_RE.sub(r"_\1", w).lower()

Demo of this regex: https://regex101.com/r/3AjsDU/1

Example usage:

>>> convert_camel_word("camelCase")
'camel_case'

>>> convert_camel_word("camelCamelCase")
'camel_camel_case'

>>> convert_camel_word("snakey_camelCase")
'snakey_camel_case'

convert_camel_word is designed to convert a single camelCase word, not strings
of arbitrary text containing camelCase words. For example:

>>> convert_camel_word("System.out.println(Arrays.toString(myArray));")
'system.out.println(_arrays.to_string(my_array));'

Instead of:

'System.out.println(Arrays.to_string(my_array));'

To convert strings of arbitrary text, the original camelCase matching from exercise 12 is used to first find
the locations of individual camelCase strings. Once found, they can be passed to
convert_camel_word and inserted into the correct position of the original text.


# factored out of existing find_camel function
def find_match_groups(s: str) -> list:
    return [MatchGroup(*m.span()) for m in CAMEL_RE.finditer(s)]


def convert_camel_line(l: str) -> str:
    # find individual camelCase strings within `l`
    for g in find_match_groups(l):
        # replace with snake_case equivalent
        l = l[0 : g.start] + convert_camel_word(l[g.start : g.end]) + l[g.end :]
    return l


def convert_camel(lines: Iterable[str]) -> Generator[str, None, None]:
    return (convert_camel_line(l) for l in lines)

Applying the conversion to a file involves backing up the original file, converting
each line in turn and writing to a new file:

def transform_camel(file: Path):
    """
    Transform all occurences of camelCase strings in `file` to snake_case. The
    original file is renamed with a ".backup" extension to prevent data loss.
    """
    original = Path(str(file) + ".backup")
    file.rename(original)
    with original.open() as source:
        with file.open("w") as dest:
            dest.writelines(convert_camel(source)

To wrap it up, the existing main entry point delegates according to whether reporting
or transformation is required and an additional --convert command line argument is added. Command line argument parsing is now performed by argparse instead of manually inspecting sys.argv:

import argparse


def main(files: List[Path], convert=False):
    for f in files:
        if convert:
            convert_camel(f)
        else:
            report_camel(f, show_filenames=len(files) > 1)


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description=sys.modules[__name__].__doc__)
    parser.add_argument("files", nargs="+", help="source files to scan for camelCase")
    parser.add_argument(
        "--convert",
        action="store_true",
        dest="convert",
        help="perform camelCase to snake_case conversion",
    )
    args = parser.parse_args()
    main([Path(f) for f in args.files], convert=args.convert)

CLI help text:

$ ./camel_finder.py -h
usage: camel_finder.py [-h] [--convert] files [files ...]

Scan source files for camelCase strings, reporting (grep style) on locations
or converting to snake_case. During conversion, orginal files are renamed with
a ".backup" extension. Yes I'm aware this can probably be achieved with a bash
one-liner.

positional arguments:
  files       source files to scan for camelCase

optional arguments:
  -h, --help  show this help message and exit
  --convert   perform camelCase to snake_case conversio

DEV Community: Ben Steadman

Reservoir Sampling

When/why is Reservoir Sampling useful?

How does Reservoir Sampling work?

How can Reservoir Sampling be implemented?

Algorithm R

Method 4 (Protection Reservoir)3

Algorithm L (Geometric Jumps)

How can a Reservoir Sampling implementation be practically tested for correctness?

Conclusion

Container with Most Water

Problem

Solution

Proof

Definitions

Loop Invariant

Termination

Python, Spark and the JVM: An overview of the PySpark Runtime Architecture

Spark vs PySpark

PySpark Execution Model

SparkContext and Py4J

Spark Workers and the PythonRDD

Summary

TPP Topic 38: Programming by Coincidence

Exercise 25

Exercise 26

Exercise 27

TPP Topic 36: Blackboards

Exercise 24

TPP Topic topic-30: Transforming Programming

Exercise 21

Exercise 22

Exercise 23

TPP Topic 29: Juggling the Real World

Exercise 19

Exercise 20

TPP Topic 26: How to Balance Resources

Challenge 1

Exercise 17

Exercise 18

Algorithm Optimisation: A Real Example

Original read_file Algorithm

Algorithmic Complexity

Identifying a Performance Problem

Improved Algorithm

Algorithmic Complexity

Performance Comparison

Conclusion

TPP Topic 25: Assertive Programming

Exercise 16

TPP Topic 23: Design by Contract

Challenge 1

Exercise 14

Exercise 15

TPP Topic 21: Text Manipulation

Exercise 11

Exercise 12

Exercise 13

Method 4 (Protection Reservoir)³

`SparkContext` and Py4J

Spark Workers and the `PythonRDD`

Original `read_file` Algorithm