Solution: where change pipe hwm, send a command (new type pipe_hwm) to peer, so peer pipe can modify hwm thread-safely