7
\$\begingroup\$

I was tasked with solving this particular problem:

A water pool which can hold a total of 800 liters of water has 3 pipes. 2 of them pour water in, the third one pours water out. 1st pipe pours 100 liters in, 2nd pipe pours 80 liters in and the third one pours 110 liters out. The pool must keep its volume between 600 and 800 liters. When the volume is greater than 700 liters, no water should be poured in, when the volume is less than 600, no water should be poured out.

This is the code I came up with:

import threading
import logging
import time
import concurrent.futures


class Pipe:
    waterVolume = 0
    def __init__(self, type):
        self.typeOfPipe = type
        self._lock = threading.Lock()

    def behaviour(self, type):
      while True:
        if type == 0:
            if Pipe.waterVolume < 700:
                with self._lock:
                    local_copy = Pipe.waterVolume
                    local_copy += 100
                    Pipe.waterVolume = local_copy
                    print('Pipe %d added 100 liters' % (self.typeOfPipe + 1))
                    print('volume is: %d' % Pipe.waterVolume)
        if type == 1:
            if Pipe.waterVolume < 700:
                with self._lock:
                    local_copy = Pipe.waterVolume
                    local_copy += 80
                    Pipe.waterVolume = local_copy
                    print('Pipe %d added 80 liters' % (self.typeOfPipe + 1))
                    print('volume is: %d' % Pipe.waterVolume)
        if type == 2:
            if Pipe.waterVolume > 600:
                with self._lock:
                    local_copy = Pipe.waterVolume
                    local_copy -= 110
                    Pipe.waterVolume = local_copy
                    print('Pipe %d removed 110 liters' % (self.typeOfPipe + 1))
                    print('volume is: %d' % Pipe.waterVolume)
        time.sleep(1.5)


if __name__ == '__main__':
    with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
        for i in range(3):
            p = Pipe(i)
            executor.submit(p.behaviour, i)

From what I've seen when satisfies the condition 600 < waterVolume < 800 for the first time, it stays there. That combined with the fact that there's no deadlocks or race conditions makes me think that this is a correct solution to the problem. Nevertheless, suggestions about improving the script are very welcome.

\$\endgroup\$
  • \$\begingroup\$ Sounds like a job for a en.wikipedia.org/wiki/PID_controller. \$\endgroup\$ – Neil Nov 6 at 20:04
  • \$\begingroup\$ looks like all your pipes are using different locks, meaning the water operations are not atomic. The lock should be common to everyone changing the water level (and encapsulated in a method that ensures no-one can just change it without taking a lock) \$\endgroup\$ – njzk2 Nov 7 at 7:00
  • \$\begingroup\$ @njzk2 so if I understand you correctly if I change the lock to be static for the Pipe class, then the threads will be synchronized? \$\endgroup\$ – Papa Grisha Nov 7 at 11:09
  • \$\begingroup\$ @PapaGrisha That is what I proposed in my answer. \$\endgroup\$ – AlexV Nov 7 at 11:28
  • 1
    \$\begingroup\$ @PapaGrisha I'm not entirely sure what part of the answer lead to this conclusion. IMHO using an RLock here does not really bring an advantage. Maybe the concept of "ownership" by a thread that would stop other threads from releasing the lock can be appealing. Since you use the lock with a context manager (with), I cannot think of a situation where that could happen. \$\endgroup\$ – AlexV Nov 7 at 12:46
7
\$\begingroup\$

Naming

type is a built-in function in Python and therefore a name to avoid. Also, the widely followed Style Guide for Python Code aka PEP 8 recommends to use lower_case_with_underscores for variable names. CamelCase or more specifically TitleCase is usually "reserved" for class names.

Duplicate code

The behaviour function has an enormous amount of duplicated code. The code code for each of the if branches is basically the same, the only variable parts are the thresholds on which those branches are triggered and the amount of water that is either drained or added. Also the type parameter of behavior is redundant, since self already knows its type. Another thing that might need a closer look is the way you use the look. My threading is a bit rusty, but if I understood your code correctly, there is a separate lock for each instance of Pipe. As a consequence, there is no real synchronization between the threads and the happily check and modify Pipe.waterVolume as they go. But maybe I'm wrong here. Under the assumption that I'm right, let's look at a refactored version of your code:

class Pipe:
    water_volume = 0
    _lock = threading.Lock()

    def __init__(self, type_, flow, activation_condition):
        self.pipe_type = type_
        self._flow = flow
        self._activation_condition = activation_condition

    @classmethod
    def pipe1(cls):
        """A pipe of type 1 adding 100l when the volume is below 700"""
        return Pipe(1, 100, lambda volume: volume < 700)

    @classmethod
    def pipe2(cls):
        """A pipe of type 2 adding 80l when the volume is below 700"""
        return Pipe(2, 80, lambda volume: volume < 700)

    @classmethod
    def pipe3(cls):
        """A pipe of type 3 draining 110l when the volume is above 600"""
        return Pipe(3, -110, lambda volume: volume > 600)

    def behaviour(self):
        while True:
            with Pipe._lock:
                if self._activation_condition(Pipe.water_volume):
                    Pipe.water_volume += self._flow
                    action = "added" if self._flow > 0 else "drained"
                    print(f"Pipe {self.pipe_type+1} {action} {abs(self._flow)} liters")
                    print(f"Volume is: {Pipe.water_volume}")
            time.sleep(1.5)

A lot has changed here, so lets go through step-by-step.

  1. I changed the naming to follow the PEP8 recommendation linked above.
  2. The lock is now shared between all instances of the class (and their associated threads) the same way the water volume is.
  3. The code duplication is gone.
    • To reach that goal, the Pipe class was made more general in that it now accepts a flow rate (positive or negative) and a condition under which this pipe is active. To have to same "preconfigured" pipes as before, I added three classmethods which act as a simple factory. Each factory method also got a short description what to expect from the returned pipe.
    • behaviour (in lack of a better idea for a name) is now quite a bit shorter since it's basically the same code for all instances which add live to add via their properties. The lock is now also acquired before checking on the current volume. From a coding standpoint that absolutely makes sense, it's however up for discussion if a real system could/would be setup with this level of synchronization. The console output is now also generically built from what is known about the pipe object currently in action. I used f-strings (available from Python 3.6+) to format the output contrary to the "old-style" % formatting. Maybe have a look at this blog post to learn more about the different kinds of string formatting available to Python programmers.

Make it stop

I don't know if you have ever heard of the poem "Der Zauberlehrling" (The Sorcerer's Apprentice) by the German writer Johann Wolfgang von Goethe. In the poem, a young sorcerer summons a broom to help him carry water in order to clean his masters house. When he is done with the task, he tries to stop the broom from carrying more and more water into the house but fails to do so. From there on the events quickly spiral out of control until his master returns and finally takes control over the broom again.

Sounds vaguely familiar? Well, that's because once one has started your program, not even Ctrl+C will stop it (at least under Windows). This answer on SO could be used as a starting point to make that work.


A few closing words on threading in Python, especially in the most widely used reference implementation CPython. CPython has something called the global interpreter lock or GIL for short. The GIL takes care that at any time there is only a single thread executing Python code. That has significant impact on what you can expect to see when changing your pure Python code to multi-threading. Positive gains are mainly found when you have blocking IO operations or something like calls to a web API where the underlying code actively releases the GIL in order to allow other threads to do their work. Apart from that, the Python interpreter will also stop threads from time to time in order to allow over threads to run. When this happens depends on which Python version you are using (Python 2 vs. Python 3).

Since this is a topic that can quite involved, let me give you another link to a blog post on realpython and to a question/answer on Stack Overflow to read more about that topic.

\$\endgroup\$
  • 3
    \$\begingroup\$ You're confusing camelCase with TitleCase (aka PascalCase). PEP-8 does not recommend any camels, but it does recommend titles for classes. And your camel uses titlecase convention. \$\endgroup\$ – Gloweye Nov 7 at 8:20

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.