[FLINK-16645] Limit the maximum backlogs in subpartitions #11567

Jiayi-Liao · 2020-03-30T13:36:21Z

What is the purpose of the change

In the case of data skew, most of the buffers in #LocalBufferPool can be requested away by one certain partition, which increases the in-flight data and slow down the buffer alignment.

To solve this problem, we add a config to limit the maximum backlog per subpartition. In this way, we can check subpartition's backlog and make the stream task unavailable if the limitation is exceeded.

Brief change log

Add taskmanager.network.max-backlogs-per-subpartition to limit the maximum backlogs per subpartition.
Make the stream unavailable / available by monitoring buffers of each subpartition in LocalBufferPool
Add class SubpartitionRecycler in LocalBufferPool as the recycler of buffer requested from subpartitions.
Add subpartition parameter version of requestSegment() and recycle(MemorySegment segment).
Rename getBufferBuilder() to getNewBufferBuilder(int targetChannel) in RecordWriter because we already have a function named getBufferBuilder(int targetChannel).

Verifying this change

testIsAvailableOrNot tests #ResultPartition#getAvailableFuture when taskmanager.network.max-backlogs-per-subpartition is not set (default value is Integer.MAX_VALUE).
testMaxBuffersPerChannelAndAvailability test #LocalBufferPool#getAvailableFuture when taskmanager.network.max-backlogs-per-subpartition is set.

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): no
The public API, i.e., is any changed class annotated with @Public(Evolving): no
The serializers: no
The runtime per-record code paths (performance sensitive): yes
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn/Mesos, ZooKeeper: no
The S3 file system connector: no

Documentation

Does this pull request introduce a new feature? yes
If yes, how is the feature documented? yes

flinkbot · 2020-03-30T13:41:41Z

Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.

Automated Checks

Last check on commit b0ea170 (Mon Mar 30 13:41:41 UTC 2020)

Warnings:

No documentation files were touched! Remember to keep the Flink docs up to date!

_{Mention the bot in a comment to re-run the automated checks.}

Review Progress

❓ 1. The [description] looks good.
❓ 2. There is [consensus] that the contribution should go into to Flink.
❓ 3. Needs [attention] from.
❓ 4. The change fits into the overall [architecture].
❓ 5. Overall code [quality] is good.

Please see the Pull Request Review Guide for a full explanation of the review process.

The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot approve description to approve one or more aspects (aspects: description, consensus, architecture and quality)
@flinkbot approve all to approve all aspects
@flinkbot approve-until architecture to approve everything until architecture
@flinkbot attention @username1 [@username2 ..] to require somebody's attention
@flinkbot disapprove architecture to remove an approval you gave earlier

flinkbot · 2020-03-30T14:01:28Z

CI report:

b25ccec Travis: SUCCESS Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run travis re-run the last Travis build
@flinkbot run azure re-run the last Azure build

pnowojski

Thanks for the contribution @Jiayi-Liao. I've left couple of comments, I think the code looks mostly ok, but I have to think more about the concurrency issue.

pnowojski · 2020-03-30T16:16:46Z

flink-core/src/main/java/org/apache/flink/configuration/NettyShuffleEnvironmentOptions.java

+	 */
+	@Documentation.Section(Documentation.Sections.ALL_TASK_MANAGER_NETWORK)
+	public static final ConfigOption<Integer> NETWORK_MAX_BACKLOGS_PER_SUBPARTITION =
+			key("taskmanager.network.max-backlogs-per-subpartition")


rename to max-buffers-per-channel to be consistent with buffers-per-channel?

flink-core/src/main/java/org/apache/flink/configuration/NettyShuffleEnvironmentOptions.java

pnowojski · 2020-03-30T16:28:14Z

flink-core/src/main/java/org/apache/flink/configuration/NettyShuffleEnvironmentOptions.java

+					.withDescription("Number of max backlogs can be used for each output subpartition." +
+							" If a subpartition exceeds the number of max backlogs, it will make the ResultPartition unavailable and" +
+							" block the processing. This benefits in reducing the in-flight data and speeding up the barrier alignment" +
+							" when most of the buffers are going to one subpartition (data skew). This limitation is not strictly" +
+							" guaranteed, which usually happens in one-to-many operators like flatmap.");
+


Number of max buffers that can be used for each channel. If a channel exceeds the number of max buffers, it will make the become unavailable, cause the back pressure and block the data processing. This might speed up checkpoint alignment by preventing excessive growth of the buffered in-flight data in case of data skew and high number of configured floating buffers. This limit is not strictly guaranteed, and can be ignored by things like flatMap operators, records spanning multiple buffers or single timer producing large amount of data.

?

pnowojski · 2020-03-30T16:36:00Z

flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartition.java

+
+	private final int maxBacklogInSubpartition;
+
+	private int unavailableSubpartitionsCnt;


unavailableSubpartitionsCnt -> unavailableSubpartitionsCount

pnowojski · 2020-03-30T16:39:38Z

flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartition.java

+	 * Check whether all subpartitions' backlogs are less than the limitation of max backlogs, and make this partition
+	 * available again if yes.
+	 */
+	@GuardedBy("buffers")


That's not true. First of all this class can not relay on an implementation detail of the PipelinedSubpartition (which is only one of two implementations of ResultSubpartition class).

Secondly each PipelinedSubpartition has it's own buffers field, so the updates here to unavailableSubpartitionsCnt are complete not synchronized. Some simple solution might be to change unavailableSubpartitionsCnt to an AtomicLong, but let me think about alternative solutions.

Another solution comes to my mind is BitSet structure and each subpartition take care of its own single bit. We can easily construct such a structure by using long[] (new long[1024] can be used by 64 * 1024 subpartitions). And instead of using a counter, we can use bit operations to check whether all long values in long[] are 0.

Update: The solution looks still not thread safe...

I would expect the smallest atomic writes/reads would be bytes, not bits. With longs, threads performing

long l = bitSet[i]; l |= 1 << n; bitSet[i] = l;

would be overriding bits from other threads sharing the same long. So you would need single byte per subpartition.

wsry · 2020-04-03T03:33:48Z

...ntime/src/main/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartition.java

@@ -342,6 +342,7 @@ private void decreaseBuffersInBacklogUnsafe(boolean isBuffer) {
 		assert Thread.holdsLock(buffers);
 		if (isBuffer) {
 			buffersInBacklog--;
+			parent.notifyDecreaseBacklog(buffersInBacklog);


the above two lines can be replaced with:
parent.notifyDecreaseBacklog(--buffersInBacklog);

wsry · 2020-04-03T03:34:06Z

...ntime/src/main/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartition.java

@@ -355,6 +356,7 @@ private void increaseBuffersInBacklog(BufferConsumer buffer) {

 		if (buffer != null && buffer.isBuffer()) {
 			buffersInBacklog++;
+			parent.notifyIncreaseBacklog(buffersInBacklog);


the above two lines can be replaced with:
parent.notifyDecreaseBacklog(++buffersInBacklog);

wsry · 2020-04-03T03:36:18Z

flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartition.java

+			if (unavailableSubpartitionsCount == 0) {
+				CompletableFuture<?> toNotify = availabilityHelper.getUnavailableToResetAvailable();
+				toNotify.complete(null);
+				int[] a = new int[1024];


why we need this?

It's an accident... removing now.

wsry · 2020-04-03T03:36:56Z

flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartition.java

+	public void notifyDecreaseBacklog(int buffersInBacklog) {
+		if (buffersInBacklog == maxBuffersPerChannel) {
+			unavailableSubpartitionsCount--;
+			if (unavailableSubpartitionsCount == 0) {


the above two lines can be replaced with:
if (--unavailableSubpartitionsCount == 0) {

wsry · 2020-04-03T03:39:13Z

flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartition.java

+	public void notifyIncreaseBacklog(int buffersInBacklog) {
+		if (buffersInBacklog == maxBuffersPerChannel + 1) {
+			unavailableSubpartitionsCount++;
+			if (unavailableSubpartitionsCount == 1) {


the above two lines can be replaced with:
if (++unavailableSubpartitionsCount == 1) {

wsry · 2020-04-03T03:44:53Z

...src/main/java/org/apache/flink/runtime/taskmanager/NettyShuffleEnvironmentConfiguration.java

@@ -74,6 +74,8 @@

 	private final String compressionCodec;

+	private final int maxBuffersPerChannel;


it's better to update the hashCode, equals and toString method

wsry · 2020-04-03T03:46:44Z

...runtime/src/test/java/org/apache/flink/runtime/io/network/partition/ResultPartitionTest.java

+
+			assertTrue(resultPartition.getAvailableFuture().isDone());
+
+			BufferAvailabilityListener listener0 = mock(BufferAvailabilityListener.class);


it's better to avoid the use of mock

You mean I should replace "mock" with a "TestBufferAvailabilityListener" here? I can do this but I wonder what the difference is.

I heard from other guys that the community is trying to avoid using mock, it's not a problem for me if there is no concern from other people.

Yes, we are discouraging using mockito for various of reasons:
https://flink.apache.org/contributing/code-style-and-quality-common.html#design-for-testability (avoid mockito section)
https://docs.google.com/presentation/d/1fZlTjOJscwmzYadPGl23aui6zopl94Mn5smG-rB0qT8/edit?usp=sharing

pnowojski · 2020-04-04T13:32:59Z

From the Travis failure It looks like you need to update docs

pnowojski · 2020-04-04T13:57:29Z

flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartition.java

+		if (buffersInBacklog == maxBuffersPerChannel) {
+			if (--unavailableSubpartitionsCount == 0) {
+				CompletableFuture<?> toNotify = availabilityHelper.getUnavailableToResetAvailable();
+				toNotify.complete(null);


this is being executed under the buffers lock. In the past we had quite a bit of problem of executing callbacks under a lock, leading to deadlocks. However since we introduced mailbox model, it might be less risky to do so, as I think the only thing this call back can do is enqueue something into the mailbox/resume default action processing? What do you think @wsry ? Could you double check if I'm not missing anything?

I think you are right, the callback only enqueues a resume mail so there should be no deadlock issue.

pnowojski · 2020-04-04T14:15:33Z

flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartition.java

+	public void notifyDecreaseBacklog(int buffersInBacklog) {
+		if (buffersInBacklog == maxBuffersPerChannel) {
+			if (--unavailableSubpartitionsCount == 0) {
+				CompletableFuture<?> toNotify = availabilityHelper.getUnavailableToResetAvailable();
+				toNotify.complete(null);
+			}
+		}
+	}
+
+	/**
+	 * Check whether any subpartition's backlog exceeds the limitation of max backlogs, and make this partition
+	 * unavailabe if yes.
+	 */
+	public void notifyIncreaseBacklog(int buffersInBacklog) {
+		if (buffersInBacklog == maxBuffersPerChannel + 1) {
+			if (++unavailableSubpartitionsCount == 1) {
+				availabilityHelper.resetUnavailable();
+			}
+		}
+	}


Continuing the previous thread

each PipelinedSubpartition has it's own buffers field, so the updates here to unavailableSubpartitionsCnt are complete not synchronized. Some simple solution might be to change unavailableSubpartitionsCnt to an AtomicLong, but let me think about alternative solutions.

There is also another issue. AvailabilityHelper is not thread safe class, while you are accessing and modifying it's state from different threads, so my previous idea with the AtomicLong unavailableSubpartitionsCount wouldn't work. This also affects getAvailableFuture.

I don't know if that can be solved in an efficient way. One thing that could help a bit, is some old idea of mine
pnowojski@24e360e
that there is no reason for each subpartition to have unique lock. Having one global per partition lock would help with couple of issues, like flushAll(), some load rebalancing ideas that we had or here. However we would have to also synchronize ResultPartition#getAvailableFuture(...) which would affect performance on the happy/hot path :/

Maybe a solution would be to not use availabilityHelper here at all. But instead availabilityHelper.resetUnavailable(); enqueue an action in the mailbox, which would suspend default action and instead of toNotify.complete(null);, also enqueue a mailbox action to re-enable default actions. That would increase a cost of #notifyIncreaseBacklog call, but not on the happy/hot path, but only if this task is being back-pressured, so when performance shouldn't matter.

However we would still need either AtomicLong or global ResultPartition.lock to not have race conditions on unavailableSubpartitionsCount. global ResultPartition.lock would be cleaner and faster solution.

(lets think this though before jumping to implementation)

Ok, I've just thought about probably a simpler and better idea. Why don't we move this anti data skew logic inside the LocalBufferPool? It's already synchronized for recycling the memory segments from netty threads and requesting from the task thread and it already implements AvailabilityProvider with an AvailabilityHelper. It currently doesn't know which channel/subpartition assigned to which buffer but it might be doable to pass this knowledge there:

add subpartition index argument to ResultPartitionWriter#getBufferBuilder

embed it into the NetworkBuffer inside LocalBufferPool#toBuffer? Maybe via passing a custom BufferRecycler?

use the subpartition id during recycling

Couple of caveats:

LocalBufferPool is also used for input gates at least, so it might need to support managing buffers with and without channels accounting.

for BroadcastRecordWriter the accounting would either need to be disabled, or Integer.MAX_VALUE could be used to indicate to account all channels?

there might be some other code paths leading to LocalBufferPool that do not care about anti data skew accounting.

Mmm.. it seems to make sense, but let me make sure I understand this correctly.
Firstly we maintain a int[] subpartitionsBufferCount in LocalBufferPool to represent the count of each subpartition's buffers.
(1) About adding buffers. We add an new version of requestMemorySegment(), maybe something like requestMemorySegment(int subpartitionIndex) and operate the subpartitionsBufferCount inside the new function, which will only be called by ResultPartition without affecting the calling stack of InputGate.
(2) About recycling. Create a custom BufferRecyler with channel information like new CustomBufferRecycler(subpartitionIndex) to make sure we know where the buffer comes from when recycling. And this means we need to create a single BufferRecycler instance for every subpartition. If I understand this part correctly, wouldn't it be adding too many (thousands of) BufferRecycler instances here?

Yes, more or less that was my last idea (keep in mind that I might be missing something).

If I understand this part correctly, wouldn't it be adding too many (thousands of) BufferRecycler instances here?

For one thing, we could create one instance of CustomBufferRecycler per subpartitionIndex and keep it referenced on some array field CustomBufferRecycler[] recyclers - we don't have to keep creating them on each request - so it shouldn't affect performance/GC pressure.

Secondly, we already have per subpartition index multiple classes (for example PipelinedSubpartition, PipelinedSubpartitionView, RemoteInputChannel, BufferBuilder, BufferConsumer and all of theirs fields + buffers/segments themselves). One more shouldn't matter?

Thirdly, CustomBufferRecycler should be ~16 bytes, that's just 160KB for 10000 channels.

Lastly, we could also hide subpartitionIndex inside NetworkBuffer class, but I think that might be less clean taking into account that InputGate won't be using this field? But I might be wrong/persuaded the other direction here.

@Jiayi-Liao just a small heads up. There is another PR that's in progress which is touching similar places (take a look at RecordWriter#getBufferBuilder in here #11564) - it's adding non blocking variant of requestingBufferBuilder from the LocalBufferPool up to the RecordWriter. It will be probably merged before your PR, so you will probably have some some conflicts during one of the rebase, that should be not that difficult to resolve, but just keep it in mind.

@pnowojski Thanks for reminding :). A new idea occurs to me this morning, maybe it'll help open our thoughts a little bit. Since each subpartition is synchronized on buffers, what if we let each subpartition know its buffers' limit and we create the availability future in every subpartition? A subpartition will be unavailable when number of buffers are increasing and exceeds the limit, will be available when number of buffers are decreasing. In ResultPartition, there'll be something like this:

public boolean addBufferConsumer(BufferConsumer bufferConsumer, int subpartitionIndex) { ... ... subpartition.add(bufferConsumer); if (!subpartition.isAvailabe()) { subpartitionsFuture.combine(subpartition.availableFuture); } ... }

But the problem here is we may have hundreds of futures combined together in the subpartitionsFuture if user sets a low value to the number of max buffers.

Yes I guess you are right.

But the problem here is we may have hundreds of futures combined together in the subpartitionsFuture if user sets a low value to the number of max buffers.

I think that wouldn't be a serious issue. Once a single sub-partition is backpressured because of data skew, data processing should pause. So as long as we are respecting the availability, there should be at most one blocked supbartition.

But it would be an issue nonetheless. It could happen for example for flatMap operator, when it's producing a lot of output on a single call.

So the subpartitionsFuture would have to maintain a List<CompletableFuture> to keep track of all of the blocked subpartitions. With this in mind, I'm not entirely sure if this approach would be easier/better than going through LocalBufferPool? Maybe it would be better to have a single source of truth for output availability (LocalBufferPool), instead of distributing the logic among LocalBufferPool and subpartitions?

Okay. I'm going to implement this in your way. This will introduce a few function changes in LocalBufferPool:

Interface changes in BufferProvider. Add subpartitionIndex as a parameter to requestBufferBuilder and requestBufferBuilderBlocking so that BufferPool knows which subpartition the request comes from.

Add requestMemorySegment(int subpartitionIndex) and recycle(int subpartitionIndex), and new function will reuse most logic of the original function, and the original one will be like this:

private MemorySegment requestMemorySegment() throws IOException { return requestMemorySegment(INVALID_SUBPARTITION_INDEX); }

Introduce class SubpartitionBufferRecycler.

class SubpartitionBufferRecycler implements BufferRecycler { private int subpartitionIndex; private LocalBufferPool bufferPool; SubpartitionBufferRecycler(int subpartitionIndex, LocalBufferPool bufferPool) { this.subpartitionIndex = subpartitionIndex; this.bufferPool = bufferPool; } @Override public void recycle(MemorySegment memorySegment) { bufferPool.recycle(memorySegment, subpartitionIndex); } }

👍

INVALID_SUBPARTITION_INDEX -> UNKNOWN_SUBPARTITION_INDEX? I guess it would be nicer to avoid using some magic constants, but to just keep requestMemorySegment() without the subpartitionIndex parameter besides a version with the parameter, but I don't see how this could be done without duplicating quite a bit of code.

👍

@pnowojski InputChannel and Subpartition share the function requestMemorySegment(), so it looks like we don't have another way except using UNKKNOW_SUBPARTITION_INDEX if we want to reuse the logic in original function. BTW, I've updated the description and the codes, perhaps you can take a look.

pnowojski

Thanks for the update @Jiayi-Liao . I think the change looks mostly good. I've left couple of smaller comments.

flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/ResultPartition.java

flink-runtime/src/main/java/org/apache/flink/runtime/io/network/api/writer/RecordWriter.java

flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/LocalBufferPool.java

flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/BufferProvider.java

flink-runtime/src/test/java/org/apache/flink/runtime/io/network/buffer/LocalBufferPoolTest.java

...runtime/src/test/java/org/apache/flink/runtime/io/network/partition/ResultPartitionTest.java

flink-runtime/src/main/java/org/apache/flink/runtime/io/network/buffer/LocalBufferPool.java

Jiayi-Liao · 2020-04-25T06:14:38Z

@pnowojski I've updated the codes according to your comments and here are the latest changes:

rename getNewBufferBuilder(int) to requestNewBufferBuilder(int).
add numSubpartitions(equals to 0 if not set) and maxBuffersPerChannel(equals to Integer.MAX_VALUE if not set) into the constructor of LocalBufferPool.
keep requestBufferBuilder() and requestBufferBuilderBlocking() and make UNKNOWN_CHANNEL only visible inside LocalBufferPool.
add unavailableSubpartitionsCount > 0 condition in LocalBufferPool#getAvailableFuture().
optimize the tests.

pnowojski

Thanks for the update @Jiayi-Liao :) Change LGTM. I've fixed couple of minor formatting issues, squashed the commits and added a proper commit message:

pnowojski@6b6c066

I'm waiting for travis run and benchmarks to complete on this commit before merging.

flink-core/src/main/java/org/apache/flink/configuration/NettyShuffleEnvironmentOptions.java

…work.memory.max-buffers-per-channel

pnowojski · 2020-04-29T06:05:28Z

Thanks for your work @Jiayi-Liao. I applied couple of white space fixes and changed the default value of config parameter to 10 merged and to master as 02978da after passing private travis build https://travis-ci.org/github/pnowojski/flink/builds/680603777

Quick glance before merging on benchmarks didn't reveal any performance regression, but we should check the long term trends.

Jiayi-Liao · 2020-04-29T07:08:12Z

@pnowojski @wsry @zhijiangW Thanks for your review!

rmetzger added the review=description? label Mar 30, 2020

rmetzger added the component=Runtime/Network label Mar 30, 2020

pnowojski self-requested a review March 30, 2020 16:14

pnowojski requested changes Mar 30, 2020

View reviewed changes

wsry reviewed Apr 3, 2020

View reviewed changes

pnowojski reviewed Apr 4, 2020

View reviewed changes

Jiayi-Liao force-pushed the FLINK-16645-limit-backlog branch from 9a8522c to 926d299 Compare April 21, 2020 15:15

Jiayi Liao added 4 commits April 21, 2020 23:36

Limit the maximum backlogs in subpartitions

f5eeeca

fix docs and rename maxBacklogsPerSubpartition to maxBuffersPerChannel

06acdbd

simplify codes and add new config in hashCode/equal/toString method

71a30db

move maxBuffersPerChannel from ResultPartition to LocalBufferPool

260dec1

pnowojski reviewed Apr 22, 2020

View reviewed changes

initialize subpartitions in constructor of LocalBufferPool

d2fbea2

Jiayi-Liao requested a review from pnowojski April 26, 2020 05:09

pnowojski approved these changes Apr 27, 2020

View reviewed changes

pnowojski reviewed Apr 27, 2020

View reviewed changes

flink-core/src/main/java/org/apache/flink/configuration/NettyShuffleEnvironmentOptions.java Outdated Show resolved Hide resolved

pnowojski reviewed Apr 27, 2020

View reviewed changes

flink-core/src/main/java/org/apache/flink/configuration/NettyShuffleEnvironmentOptions.java Show resolved Hide resolved

rename taskmanager.network.max-buffers-per-channel to taskmanager.net…

b25ccec

…work.memory.max-buffers-per-channel

pnowojski closed this Apr 29, 2020

Jiayi-Liao deleted the FLINK-16645-limit-backlog branch May 23, 2020 14:23


		private final int maxBacklogInSubpartition;

		private int unavailableSubpartitionsCnt;

		@@ -74,6 +74,8 @@

		private final String compressionCodec;

		private final int maxBuffersPerChannel;


		assertTrue(resultPartition.getAvailableFuture().isDone());

		BufferAvailabilityListener listener0 = mock(BufferAvailabilityListener.class);

[FLINK-16645] Limit the maximum backlogs in subpartitions #11567

[FLINK-16645] Limit the maximum backlogs in subpartitions #11567

Conversation

Jiayi-Liao commented Mar 30, 2020 • edited

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

flinkbot commented Mar 30, 2020

Automated Checks

Review Progress

flinkbot commented Mar 30, 2020 • edited

CI report:

pnowojski left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Jiayi-Liao Mar 31, 2020 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

pnowojski commented Apr 4, 2020

pnowojski Apr 4, 2020 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

pnowojski Apr 4, 2020 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Jiayi-Liao Apr 5, 2020 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

pnowojski Apr 20, 2020 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

pnowojski left a comment

Choose a reason for hiding this comment

Jiayi-Liao commented Apr 25, 2020 • edited

pnowojski left a comment • edited

Choose a reason for hiding this comment

pnowojski commented Apr 29, 2020

Jiayi-Liao commented Apr 29, 2020

Jiayi-Liao commented Mar 30, 2020 •

edited

flinkbot commented Mar 30, 2020 •

edited

Jiayi-Liao Mar 31, 2020 •

edited

pnowojski Apr 4, 2020 •

edited

pnowojski Apr 4, 2020 •

edited

Jiayi-Liao Apr 5, 2020 •

edited

pnowojski Apr 20, 2020 •

edited

Jiayi-Liao commented Apr 25, 2020 •

edited

pnowojski left a comment •

edited