Port ParDo, GBK, etc. from other rust project.

[robertwb]

Also fix some bugs in construction and add tests.

---

Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:

• Choose reviewer(s) and mention them in a comment (R: @username).
• Mention the appropriate issue in your description (for example: addresses #123), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, comment fixes #<ISSUE NUMBER> instead.
• Update CHANGES.md with noteworthy changes.
• If this contribution is large, please file an Apache Individual Contributor License Agreement.

See the Contributor Guide for more tips on how to make review process smoother.

To check the build health, please visit https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md

GitHub Actions Tests Status (on master branch)

See CI.md for more information about GitHub Actions CI.

[github-actions]

This pull request has been marked as stale due to 60 days of inactivity. It will be closed in 1 week if no further activity occurs. If you think that’s incorrect or this pull request requires a review, please simply write any comment. If closed, you can revive the PR at any time and @mention a reviewer or discuss it on the dev@beam.apache.org list. Thank you for your contributions.

[laysakura]

@robertwb Hi. I'm trying to develop Rust SDK with you guys!
I tested the HEAD of the topic branch ( eb3894d ) and found probabilistic test failures.

How to reproduce

while cargo test -- --skip target/debug ; do echo '' ; done

Failure examples

running 9 tests
test tests::coders_test::tests::test_general_object_coder ... ok
test tests::primitives_test::tests::run_impulse_expansion ... ok
test tests::primitives_test::tests::run_direct_runner ... ok
test tests::worker_test::tests::test_operator_construction ... ok
test tests::primitives_test::tests::run_flatten ... ok
test tests::primitives_test::tests::run_map ... ok
test tests::primitives_test::tests::ensure_assert_fails - should panic ... ok
test tests::primitives_test::tests::run_gbk ... FAILED
test tests::coders_test::tests::test_standard_coders ... ok

failures:

---- tests::primitives_test::tests::run_gbk stdout ----
thread 'tests::primitives_test::tests::run_gbk' panicked at 'called `Option::unwrap()` on a `None` value', src/internals/serialize.rs:79:69

running 9 tests
test tests::coders_test::tests::test_general_object_coder ... ok
test tests::primitives_test::tests::run_direct_runner ... ok
test tests::primitives_test::tests::run_impulse_expansion ... ok
test tests::primitives_test::tests::run_map ... FAILED
test tests::primitives_test::tests::ensure_assert_fails - should panic ... ok
test tests::primitives_test::tests::run_gbk ... ok
test tests::primitives_test::tests::run_flatten ... ok
test tests::worker_test::tests::test_operator_construction ... ok
test tests::coders_test::tests::test_standard_coders ... ok

failures:

---- tests::primitives_test::tests::run_map stdout ----
thread 'tests::primitives_test::tests::run_map' panicked at 'called `Option::unwrap()` on a `None` value', src/internals/serialize.rs:110:57
note: run with `RUST_BACKTRACE=1` environment variable to display a backtrace

running 9 tests
test tests::coders_test::tests::test_general_object_coder ... ok
test tests::primitives_test::tests::run_direct_runner ... ok
test tests::primitives_test::tests::run_impulse_expansion ... ok
test tests::primitives_test::tests::ensure_assert_fails - should panic ... ok
test tests::primitives_test::tests::run_gbk ... ok
test tests::primitives_test::tests::run_flatten ... FAILED
test tests::primitives_test::tests::run_map ... FAILED
test tests::worker_test::tests::test_operator_construction ... FAILED
test tests::coders_test::tests::test_standard_coders ... ok

failures:

---- tests::primitives_test::tests::run_flatten stdout ----
thread 'tests::primitives_test::tests::run_flatten' panicked at 'called `Option::unwrap()` on a `None` value', src/worker/operators.rs:575:10

---- tests::primitives_test::tests::run_map stdout ----
thread 'tests::primitives_test::tests::run_map' panicked at 'called `Result::unwrap()` on an `Err` value: PoisonError { .. }', src/worker/operators.rs:172:52

---- tests::worker_test::tests::test_operator_construction stdout ----
thread 'tests::worker_test::tests::test_operator_construction' panicked at 'called `Result::unwrap()` on an `Err` value: PoisonError { .. }', src/worker/operators.rs:172:52

[sjvanrossum]

My guess here is that this is caused by two things.
The lock on OPERATORS_BY_URN (which seems unnecessary since it's only read) is getting poisoned when deserialize_fn is called on incompatible types.
That's the core of the issue, which is caused by locking and releasing twice in serialize_fn while storing the length of the map to be used as the name of the entry.

That map needs to be replaced anyway, since it would be empty if you were to run pipeline construction and execution in separate processes or machines. I'm thinking of replacing that with bincode, serde and typetag for trait objects, but let's not dwell on that for too long right now.

[sjvanrossum]

@nivaldoh @laysakura Merge this first and then have a look at the other PRs?

[dahlbaek]

@sjvanrossum in case you're interested, the race was removed in laysakura#7 and related to the mutex as you expected.

[github-actions]

This pull request has been marked as stale due to 60 days of inactivity. It will be closed in 1 week if no further activity occurs. If you think that’s incorrect or this pull request requires a review, please simply write any comment. If closed, you can revive the PR at any time and @mention a reviewer or discuss it on the dev@beam.apache.org list. Thank you for your contributions.

[github-actions]

This pull request has been closed due to lack of activity. If you think that is incorrect, or the pull request requires review, you can revive the PR at any time.