New issue

Jump to bottom

PROTON-1442: [Cpp] Support for local transactions #437

Draft

DreamPearl wants to merge 14 commits into apache:main from DreamPearl:local-transactions

+968 −22

Contributor

DreamPearl commented Nov 20, 2024 •

edited

Loading

AMQP Transaction Sequence:

Declare transaction:

Client establishes link to Broker (Transaction resource) to target with transaction coordinator type (usual types are sender/receiver) (ATTACH frame)
Client (Transaction Controller) sends a special message to that link to create transaction (TRANSFER frame)
Broker returns a disposition with the transaction id (DISPOSITION frame)

Send Message in the transaction:

Client sends a message to the broker with transaction id in the state field.

Commit/Abort

DreamPearl marked this pull request as draft

November 20, 2024 08:57

DreamPearl force-pushed the local-transactions branch 2 times, most recently from 74818fc to eb4514b Compare

November 21, 2024 18:28

DreamPearl force-pushed the local-transactions branch from 3e96834 to fa9236d Compare

December 3, 2024 12:56

astitcher requested changes

View reviewed changes

Member

astitcher left a comment

I've taken a fairly thorough look. If you have any questions about these review items we should discuss further.

cpp/include/proton/container.hpp Outdated Show resolved Hide resolved

cpp/include/proton/target_options.hpp

Comment on lines +91 to +92

		PN_CPP_EXTERN target_options& type(const int);

Member

astitcher Dec 4, 2024

This should not be needed. Introduce a new coordinator class that is peer to sender and receiver

cpp/include/proton/tracker.hpp Outdated Show resolved Hide resolved

cpp/include/proton/transaction.hpp Outdated Show resolved Hide resolved

cpp/include/proton/transaction.hpp Outdated

Comment on lines 101 to 62

+              class
+              PN_CPP_CLASS_EXTERN transaction_handler {
+                public:
+                  PN_CPP_EXTERN virtual ~transaction_handler();
+                  /// Called when a local transaction is declared.
+                  PN_CPP_EXTERN virtual void on_transaction_declared(transaction);
+                  /// Called when a local transaction is discharged successfully.
+                  PN_CPP_EXTERN virtual void on_transaction_committed(transaction);
+                  /// Called when a local transaction is discharged unsuccessfully (aborted).
+                  PN_CPP_EXTERN virtual void on_transaction_aborted(transaction);
+                  /// Called when a local transaction declare fails.
+                  PN_CPP_EXTERN virtual void on_transaction_declare_failed(transaction);
+                  /// Called when the commit of a local transaction fails.
+                  PN_CPP_EXTERN virtual void on_transaction_commit_failed(transaction);
+              };
+              } // namespace proton

Member

astitcher Dec 4, 2024

I think this should be in a different header file like message and messaging_handler

Member

astitcher Dec 18, 2024

So if the API makes transactions hidden in a session then these callbacks either go away or instead become session callbacks: on_session_transaction_committed(session&), etc. I think transaction_declared() goes away entirely and it's purpose is now another use for on_session_open(session&). on_.._declare_failed should be handled by the on_session_error(session&).

cpp/include/proton/transaction.hpp Outdated Show resolved Hide resolved

cpp/include/proton/transaction.hpp Outdated Show resolved Hide resolved

cpp/include/proton/transaction.hpp Outdated Show resolved Hide resolved

cpp/examples/tx_recv.cpp Outdated Show resolved Hide resolved

cpp/examples/tx_recv.cpp Outdated Show resolved Hide resolved

DreamPearl force-pushed the local-transactions branch from fa9236d to 347cf6e Compare

December 16, 2024 13:29

astitcher requested changes

View reviewed changes

Member

astitcher left a comment

I've not reviewed this comprehensively, but I think I'm now very much leaning towards not making the transaction class visible to the API user at all and having the transaction methods on the session. This is essentially the API in JMS and CMS.

cpp/include/proton/transaction.hpp Outdated Show resolved Hide resolved

cpp/include/proton/transaction.hpp Outdated Show resolved Hide resolved

cpp/include/proton/transaction.hpp Outdated Show resolved Hide resolved

cpp/include/proton/transaction.hpp Outdated

Comment on lines 101 to 62

+              class
+              PN_CPP_CLASS_EXTERN transaction_handler {
+                public:
+                  PN_CPP_EXTERN virtual ~transaction_handler();
+                  /// Called when a local transaction is declared.
+                  PN_CPP_EXTERN virtual void on_transaction_declared(transaction);
+                  /// Called when a local transaction is discharged successfully.
+                  PN_CPP_EXTERN virtual void on_transaction_committed(transaction);
+                  /// Called when a local transaction is discharged unsuccessfully (aborted).
+                  PN_CPP_EXTERN virtual void on_transaction_aborted(transaction);
+                  /// Called when a local transaction declare fails.
+                  PN_CPP_EXTERN virtual void on_transaction_declare_failed(transaction);
+                  /// Called when the commit of a local transaction fails.
+                  PN_CPP_EXTERN virtual void on_transaction_commit_failed(transaction);
+              };
+              } // namespace proton

Member

astitcher Dec 18, 2024

So if the API makes transactions hidden in a session then these callbacks either go away or instead become session callbacks: on_session_transaction_committed(session&), etc. I think transaction_declared() goes away entirely and it's purpose is now another use for on_session_open(session&). on_.._declare_failed should be handled by the on_session_error(session&).

cpp/include/proton/transfer.hpp Outdated

Comment on lines 98 to 100

		PN_CPP_EXTERN void transaction(transaction t);

		PN_CPP_EXTERN class transaction transaction() const;

Member

astitcher Dec 18, 2024

I think these are not use visible. They are only used internally in the implementation - especially if the transaction is hidden inside the session.

pematous reviewed

View reviewed changes

cpp/examples/tx_send.cpp

+                      send(s);
+                  }
+                  void send(proton::sender &s) {

pematous Jan 21, 2025

sender is defined as tx_recv class attribute, so I believe we don't need to pass the sender to send method.

pematous reviewed

View reviewed changes

cpp/examples/tx_recv.cpp Outdated

+                      transaction.accept(d);
+                      current_batch += 1;
+                      if(current_batch == batch_size) {
+                          transaction = proton::transaction(); // null

pematous Jan 21, 2025

I believe we should do rather commit here, that way it works as expected and the receiver is closed after expected number of messages received (with the current implementation it's not).

pematous Feb 5, 2025 •

edited

Loading

I believe we should do rather commit here, that way it works as expected and the receiver is closed after expected number of messages received (with the current implementation it's not).

above mentioned change:

-            transaction = proton::transaction(); // null
+           transaction.commit();

works as expected (against Artemis broker) when using multicast. However, when a pre-defined anycast queue is used, the commit() call ends with client segmentation fault.

pematous reviewed

View reviewed changes

cpp/examples/tx_recv.cpp

+              int main(int argc, char **argv) {
+                  std::string address("127.0.0.1:5672/examples");
+                  int message_count = 9;

pematous Jan 21, 2025

I would expect the defaults to be the same for tx_send and tx_recv (currently tx_send defaults to 6 messages while tx_recv to 9)

pematous reviewed

View reviewed changes

cpp/examples/tx_send.cpp

+                          if(current_batch == batch_size)
+                          {
+                              std::cout << " >> Txn attempt commit" << std::endl;
+                              if (batch_index % 2 == 0) {

pematous Jan 24, 2025

I see that every second batch is aborted. I'm not sure if that's intended, though it works as expected (python example is not doing any aborts).

pematous reviewed

View reviewed changes

cpp/examples/tx_send.cpp

+                      }
+                  }
+                  void on_tracker_accept(proton::tracker &t) override {

pematous Jan 24, 2025

I believe on_tracker_accept method is extra and is never fired. If so, counting confirmed doesn't make much sense (confirmed variable is not used to anything btw).

pematous reviewed

View reviewed changes

cpp/examples/tx_recv.cpp Outdated

+                  void on_message(proton::delivery &d, proton::message &msg) override {
+                      std::cout<<"# MESSAGE: " << msg.id() <<": "  << msg.body() << std::endl;
+                      transaction.accept(d);

pematous Jan 24, 2025

btw. I'm thinking if transaction.accept() does actually have any effect? I mean, when I comment it out, the program behaves the same. btw. is this transaction accept any different from delivery accept? looks both settles the messages related to the delivery.

Does explicit delivery settling play role in transaction mode? or are they mutually exclusive and only commits and aborts applies? Can ie. single message in transaction be rejected while other messages accepted?

I tried to accept / reject the delivery and it seems to have no effect in transaction mode (while in python, doing so makes the messages to be threat outside of the transaction, just like a normal messages, https://issues.redhat.com/browse/ENTMQCL-513).

astitcher and others added 14 commits

March 17, 2025 23:24


          WIP: Python work for better transaction support

3ac0c72

* Added an extra handler to the python binding so that we can handle
  transactioned dispositions


          WIP: Python example broker work for better transaction support

882a7d2

* Modified the Python example broker so that it understands transaction
  requests, prints some useful output about what is happening, but
  doesn't honor the transaction semantics. It will queue up transactioned
  messages immediately and also doesn't correctly handle outgoing
  message releases (but it doesn't for non-transactioned messages either


          Implement local transactions

cf1c847


          WIP: Fix up to transactions work:

d353cd6

* Make it compile
* Make it fit the existing software structure better


          Make example compile

343492a


          Add txn id in disposition frame bits

2d630ac


          Add transaction commit and abort functionalities

fc90aaa


          Add tx_recv file

d7bfb27


          Move the transaction_impl class logic to cpp file

22ea517


          Has errors

8f36a12


          Add coordinator class

f8eff19


          Allow multiple transactions but one at a time

cca407b


          Move txn_impl into CPP file

2d3828f


          Clean up the code

eb8afbd

DreamPearl force-pushed the local-transactions branch from 4c795ef to eb8afbd Compare

March 17, 2025 17:54

pematous reviewed

View reviewed changes

cpp/src/messaging_adapter.cpp

               void on_link_remote_open(messaging_handler& handler, pn_event_t* event) {
                   auto lnk = pn_event_link(event);
-                  // Currently don't implement (transaction) coordinator
+                  int type = pn_terminus_get_type(pn_link_remote_target(lnk));

pematous Mar 18, 2025

unused variable ‘type’

astitcher requested changes

View reviewed changes

Member

astitcher left a comment

Lopts of detailed comments here, but the main change I'd like to see is either make coordinator a subclass of target or fold it into a simple boolean option of target.

cpp/examples/tx_recv.cpp

+                  int current_batch = 0;
+                  int committed = 0;
+                  proton::session session;

Member

astitcher Mar 20, 2025

Not sure we need to store the session anywhere - I think it can always be derived inside any callback

cpp/examples/tx_recv.cpp

+                  }
+                  void on_session_open(proton::session &s) override {
+                      session = s;

Member

astitcher Mar 20, 2025

Probably not necessary

cpp/examples/tx_recv.cpp

+                  void on_message(proton::delivery &d, proton::message &msg) override {
+                      std::cout<<"# MESSAGE: " << msg.id() <<": "  << msg.body() << std::endl;
+                      session.txn_accept(d);

Member

astitcher Mar 20, 2025

session == d.session()

cpp/examples/tx_recv.cpp

+                          s.connection().close();
+                      }
+                      else {
+                          session.declare_transaction(*this);

Member

astitcher Mar 20, 2025

session == s

cpp/examples/tx_send.cpp

+                  int committed = 0;
+                  int confirmed = 0;
+                  proton::session session;

Member

astitcher Mar 20, 2025

As tx_recv: I don't think you need to store away the session.

cpp/src/proton_bits.hpp

Comment on lines +122 to +124

+              template <> struct wrapper<pn_disposition_t> {
+                  typedef disposition type;
+              };

Member

astitcher Mar 20, 2025

Inconsistent formatting

cpp/src/sender_options.cpp

@@ @@ -66,6 +66,7 @@ class sender_options::impl { @@
                   option<bool> auto_settle;
                   option<source_options> source;
                   option<target_options> target;
+                  option<coordinator_options> coordinator;

Member

astitcher Mar 20, 2025

Not needed if coordinator is a kind of target

cpp/src/sender_options.cpp

Comment on lines +86 to +89

+                          if (coordinator.set) {
+                              proton::coordinator local_t(make_wrapper<proton::coordinator>(pn_link_target(unwrap(s))));
+                              coordinator.value.apply(local_t);
+                          }

Member

astitcher Mar 20, 2025

Ditto

cpp/src/sender_options.cpp

               sender_options& sender_options::auto_settle(bool b) {impl_->auto_settle = b; return *this; }
               sender_options& sender_options::source(const source_options &s) {impl_->source = s; return *this; }
               sender_options& sender_options::target(const target_options &s) {impl_->target = s; return *this; }
+              sender_options& sender_options::coordinator(const coordinator_options &s) {impl_->coordinator = s; return *this; }

Member

astitcher Mar 20, 2025

Ditto

cpp/src/transaction.cpp

		@@ -0,0 +1,44 @@
		/*

Member

astitcher Mar 20, 2025

Call this file transaction_handler.cpp

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet