diff --git a/CMakeLists.txt b/CMakeLists.txt index 26979c3f..26149465 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -94,6 +94,12 @@ configure_package_config_file( PATH_VARS CONFIG_INC_DIR CONFIG_LIB_DIR ) +write_basic_package_version_file( + ${CMAKE_CURRENT_BINARY_DIR}/${PROJECT_NAME}-config-version.cmake + VERSION ${WORKFLOW_VERSION} + COMPATIBILITY AnyNewerVersion +) + install( FILES ${CMAKE_CONFIG_INSTALL_FILE} DESTINATION ${CMAKE_CONFIG_INSTALL_DIR} @@ -101,6 +107,12 @@ install( RENAME ${PROJECT_NAME}-config.cmake ) +install( + FILES ${CMAKE_CURRENT_BINARY_DIR}/${PROJECT_NAME}-config-version.cmake + DESTINATION ${CMAKE_CONFIG_INSTALL_DIR} + COMPONENT devel +) + install( FILES ${INCLUDE_HEADERS} ${INCLUDE_KERNEL_HEADERS} DESTINATION ${CMAKE_INSTALL_INCLUDEDIR}/${PROJECT_NAME} diff --git a/README.md b/README.md index 6e39c4c1..187fbc47 100644 --- a/README.md +++ b/README.md @@ -61,6 +61,33 @@ cd tutorial make ~~~~ +### Get started (Debian Linux): +Sogou C++ Workflow has been packaged for Debian. It is currently in Debian sid (unstable) but will eventually be placed into the stable repository. + +In order to access the unstable repository, you will need to edit your /etc/apt/sources.list file. + +Simply add the 'unstable' sub branch to your repo: +~~~~sh +deb http://deb.debian.org/ main contrib non-free +--> +deb http://deb.debian.org/ unstable main contrib non-free +~~~~ + +Once that is added, update your repo list and then you should be able to install it: +~~~~sh +sudo apt-get update +~~~~ + +To install the Workflow library for development purposes: +~~~~sh +sudo apt-get install libworkflow-dev +~~~~ + +To install the Workflow library for deployment: +~~~~sh +sudo apt-get install libworkflow1 +~~~~ + # Tutorials * Client diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt index 654851b5..214c430b 100644 --- a/src/CMakeLists.txt +++ b/src/CMakeLists.txt @@ -55,6 +55,10 @@ add_library( $ ) +if (APPLE) + target_link_libraries(${SHARED_LIB_NAME} OpenSSL::SSL OpenSSL::Crypto pthread) +endif () + set_target_properties(${STATIC_LIB_NAME} PROPERTIES OUTPUT_NAME ${PROJECT_NAME}) set_target_properties(${SHARED_LIB_NAME} PROPERTIES OUTPUT_NAME ${PROJECT_NAME}) @@ -64,7 +68,6 @@ if (KAFKA STREQUAL "y") add_dependencies(protocol_kafka LINK_HEADERS) add_dependencies(factory_kafka LINK_HEADERS) set(KAFKA_STATIC_LIB_NAME "wfkafka-static") - set(KAFKA_SHARED_LIB_NAME "wfkafka-shared") add_library( ${KAFKA_STATIC_LIB_NAME} STATIC $ @@ -72,15 +75,18 @@ if (KAFKA STREQUAL "y") $ $ ) - add_library( - ${KAFKA_SHARED_LIB_NAME} SHARED - $ - $ - $ - $ - ) set_target_properties(${KAFKA_STATIC_LIB_NAME} PROPERTIES OUTPUT_NAME "wfkafka") - set_target_properties(${KAFKA_SHARED_LIB_NAME} PROPERTIES OUTPUT_NAME "wfkafka") + if (NOT APPLE) + set(KAFKA_SHARED_LIB_NAME "wfkafka-shared") + add_library( + ${KAFKA_SHARED_LIB_NAME} SHARED + $ + $ + $ + $ + ) + set_target_properties(${KAFKA_SHARED_LIB_NAME} PROPERTIES OUTPUT_NAME "wfkafka") + endif () endif () install( @@ -92,10 +98,18 @@ install( if (KAFKA STREQUAL "y") install( - TARGETS ${KAFKA_STATIC_LIB_NAME} ${KAFKA_SHARED_LIB_NAME} + TARGETS ${KAFKA_STATIC_LIB_NAME} ARCHIVE DESTINATION ${CMAKE_INSTALL_LIBDIR} COMPONENT devel ) + if (NOT APPLE) + install( + TARGETS ${KAFKA_SHARED_LIB_NAME} + ARCHIVE + DESTINATION ${CMAKE_INSTALL_LIBDIR} + COMPONENT devel + ) + endif () endif () diff --git a/src/client/WFKafkaClient.cc b/src/client/WFKafkaClient.cc index f4b5b6bb..a8b426da 100644 --- a/src/client/WFKafkaClient.cc +++ b/src/client/WFKafkaClient.cc @@ -125,7 +125,7 @@ public: this->broker_list = new KafkaBrokerList; this->lock_status = new KafkaLockStatus; this->broker_map = new KafkaBrokerMap; - this->meta_map = new std::map; + this->meta_map = new std::map; } ~KafkaMember() @@ -149,7 +149,7 @@ public: KafkaBrokerList *broker_list; KafkaBrokerMap *broker_map; KafkaLockStatus *lock_status; - std::map *meta_map; + std::map *meta_map; private: std::atomic *ref; @@ -325,7 +325,8 @@ private: int get_node_id(const KafkaToppar *toppar); - MetaStatus get_meta_status(); + enum MetaStatus get_meta_status(); + void set_meta_status(enum MetaStatus status); std::string get_userinfo() { return this->userinfo; } @@ -673,12 +674,19 @@ void ComplexKafkaTask::kafka_parallel_callback(const ParallelWork *pwork) t->error = 0; std::pair *state_error; - + bool flag = false; for (size_t i = 0; i < pwork->size(); i++) { state_error = (std::pair *)pwork->series_at(i)->get_context(); if (state_error->first != WFT_STATE_SUCCESS) { + if (!flag) + { + flag = true; + t->lock_status.get_mutex()->lock(); + t->set_meta_status(META_UNINIT); + t->lock_status.get_mutex()->unlock(); + } t->state = state_error->first; t->error = state_error->second; } @@ -698,7 +706,7 @@ void ComplexKafkaTask::kafka_process_toppar_offset(KafkaToppar *task_toppar) if (strcmp(toppar->get_topic(), task_toppar->get_topic()) == 0 && toppar->get_partition() == task_toppar->get_partition()) { - if (task_toppar->get_error() == KAFKA_NONE && + if (task_toppar->get_error() == KAFKA_NONE && !task_toppar->reach_high_watermark()) toppar->set_offset(task_toppar->get_offset() + 1); else @@ -781,11 +789,11 @@ void ComplexKafkaTask::parse_query() } } -MetaStatus ComplexKafkaTask::get_meta_status() +enum MetaStatus ComplexKafkaTask::get_meta_status() { this->meta_list.rewind(); KafkaMeta *meta; - MetaStatus ret = META_INITED; + enum MetaStatus ret = META_INITED; while ((meta = this->meta_list.get_next()) != NULL) { switch((*this->client->member->meta_map)[meta->get_topic()]) @@ -808,6 +816,14 @@ MetaStatus ComplexKafkaTask::get_meta_status() return ret; } +void ComplexKafkaTask::set_meta_status(enum MetaStatus status) +{ + this->client_meta_list.rewind(); + KafkaMeta *meta; + while ((meta = this->client_meta_list.get_next()) != NULL) + (*this->client->member->meta_map)[meta->get_topic()] = status; +} + void ComplexKafkaTask::dispatch() { __WFKafkaTask *task; diff --git a/src/factory/KafkaTaskImpl.cc b/src/factory/KafkaTaskImpl.cc index 4f743a96..62c20d41 100644 --- a/src/factory/KafkaTaskImpl.cc +++ b/src/factory/KafkaTaskImpl.cc @@ -432,6 +432,7 @@ bool __ComplexKafkaTask::process_join_group() if (msg->get_cgroup()->get_error() == KAFKA_MISSING_TOPIC) { this->get_req()->set_api_type(Kafka_Metadata); + this->get_req()->set_alien(); update_metadata_ = true; } else if (msg->get_cgroup()->get_error() == KAFKA_MEMBER_ID_REQUIRED) @@ -478,6 +479,7 @@ bool __ComplexKafkaTask::process_metadata() { KafkaCgroup *cgroup = msg->get_cgroup(); if (cgroup->run_assignor(msg->get_meta_list(), + msg->get_alien_meta_list(), cgroup->get_protocol_name()) < 0) { this->error = errno; diff --git a/src/protocol/HttpMessage.cc b/src/protocol/HttpMessage.cc index cf1961b8..6094cde0 100644 --- a/src/protocol/HttpMessage.cc +++ b/src/protocol/HttpMessage.cc @@ -67,30 +67,6 @@ bool HttpMessage::append_output_body_nocopy(const void *buf, size_t size) return false; } -bool HttpMessage::get_output_body_nocopy(struct iovec iov[], int *iovcnt) const -{ - struct HttpMessageBlock *block; - struct list_head *pos; - int i = 0; - - list_for_each(pos, &this->output_body) - { - if (i >= *iovcnt) - { - errno = ENOBUFS; - return false; - } - - block = list_entry(pos, struct HttpMessageBlock, list); - iov[i].iov_base = (void *)block->ptr; - iov[i].iov_len = block->size; - i++; - } - - *iovcnt = i; - return true; -} - void HttpMessage::clear_output_body() { struct HttpMessageBlock *block; diff --git a/src/protocol/HttpMessage.h b/src/protocol/HttpMessage.h index cd147e14..07461b50 100644 --- a/src/protocol/HttpMessage.h +++ b/src/protocol/HttpMessage.h @@ -19,7 +19,6 @@ #ifndef _HTTPMESSAGE_H_ #define _HTTPMESSAGE_H_ -#include #include #include #include @@ -128,7 +127,6 @@ public: * msg->append_output_body_nocopy(body, size); */ bool append_output_body(const void *buf, size_t size); bool append_output_body_nocopy(const void *buf, size_t size); - bool get_output_body_nocopy(struct iovec iov[], int *iovcnt) const; void clear_output_body(); size_t get_output_body_size() const diff --git a/src/protocol/KafkaDataTypes.cc b/src/protocol/KafkaDataTypes.cc index 186d288c..0689327c 100644 --- a/src/protocol/KafkaDataTypes.cc +++ b/src/protocol/KafkaDataTypes.cc @@ -208,10 +208,9 @@ bool KafkaMeta::create_partitions(int partition_cnt) return true; } -int KafkaCgroup::run_assignor(KafkaMetaList *meta_list, - const char *protocol_name) +void KafkaCgroup::add_subscriber(KafkaMetaList *meta_list, + std::vector *subscribers) { - std::vector subscribers; meta_list->rewind(); KafkaMeta *meta; @@ -241,13 +240,21 @@ int KafkaCgroup::run_assignor(KafkaMetaList *meta_list, } if (!subscriber.get_member()->empty()) - subscribers.emplace_back(subscriber); + subscribers->emplace_back(subscriber); } +} + +int KafkaCgroup::run_assignor(KafkaMetaList *meta_list, + KafkaMetaList *alien_meta_list, + const char *protocol_name) +{ + std::vector subscribers; + this->add_subscriber(meta_list, &subscribers); + this->add_subscriber(alien_meta_list, &subscribers); struct list_head *pos; kafka_group_protocol_t *protocol; bool flag = false; - list_for_each(pos, this->get_group_protocol()) { protocol = list_entry(pos, kafka_group_protocol_t, list); diff --git a/src/protocol/KafkaDataTypes.h b/src/protocol/KafkaDataTypes.h index 44fc355c..0cc31f10 100644 --- a/src/protocol/KafkaDataTypes.h +++ b/src/protocol/KafkaDataTypes.h @@ -1255,6 +1255,44 @@ private: friend const KafkaMeta *get_meta(const char *topic, KafkaMetaList *meta_list); }; +class KafkaMetaSubscriber +{ +public: + void set_meta(KafkaMeta *meta) + { + this->meta = meta; + } + + const KafkaMeta *get_meta() const + { + return this->meta; + } + + void add_member(kafka_member_t *member) + { + this->member_vec.push_back(member); + } + + const std::vector *get_member() const + { + return &this->member_vec; + } + + static bool cmp(const kafka_member_t *m1, const kafka_member_t *m2) + { + return strcmp(m1->member_id, m2->member_id) < 0; + } + + void sort_by_member() + { + std::sort(this->member_vec.begin(), this->member_vec.end(), cmp); + } + +private: + KafkaMeta *meta; + std::vector member_vec; +}; + class KafkaCgroup { public: @@ -1353,7 +1391,11 @@ public: return this->coordinator; } - int run_assignor(KafkaMetaList *meta_list, const char *protocol_name); + int run_assignor(KafkaMetaList *meta_list, KafkaMetaList *alien_meta_list, + const char *protocol_name); + + void add_subscriber(KafkaMetaList *meta_list, + std::vector *subscribers); static int kafka_range_assignor(kafka_member_t **members, int member_elements, @@ -1466,44 +1508,6 @@ private: friend class KafkaList; }; -class KafkaMetaSubscriber -{ -public: - void set_meta(KafkaMeta *meta) - { - this->meta = meta; - } - - const KafkaMeta *get_meta() const - { - return this->meta; - } - - void add_member(kafka_member_t *member) - { - this->member_vec.push_back(member); - } - - const std::vector *get_member() const - { - return &this->member_vec; - } - - static bool cmp(const kafka_member_t *m1, const kafka_member_t *m2) - { - return strcmp(m1->member_id, m2->member_id) < 0; - } - - void sort_by_member() - { - std::sort(this->member_vec.begin(), this->member_vec.end(), cmp); - } - -private: - KafkaMeta *meta; - std::vector member_vec; -}; - class KafkaBuffer { public: diff --git a/src/protocol/KafkaMessage.cc b/src/protocol/KafkaMessage.cc index 6f2e58b5..ebc87b7d 100644 --- a/src/protocol/KafkaMessage.cc +++ b/src/protocol/KafkaMessage.cc @@ -44,7 +44,7 @@ namespace protocol #define CHECK_RET(exp) if (exp < 0) return exp #ifndef htonll -static int64_t htonll(int64_t x) +static uint64_t htonll(uint64_t x) { if (1 == htonl(1)) return x; @@ -1614,6 +1614,7 @@ KafkaMessage::KafkaMessage() this->stream = new EncodeStream; this->api_type = Kafka_Unknown; this->cur_size = 0; + this->alien = false; } KafkaMessage::~KafkaMessage() @@ -2431,11 +2432,17 @@ int KafkaRequest::encode_metadata(struct iovec vectors[], int max) else append_i32(this->msgbuf, 0); - this->meta_list.rewind(); + KafkaMetaList *p_meta_list; + if (this->alien) + p_meta_list = &this->alien_meta_list; + else + p_meta_list = &this->meta_list; + + p_meta_list->rewind(); KafkaMeta *meta; int topic_cnt = 0; - while ((meta = this->meta_list.get_next()) != NULL) + while ((meta = p_meta_list->get_next()) != NULL) { append_string(this->msgbuf, meta->get_topic()); ++topic_cnt; @@ -3053,8 +3060,14 @@ int KafkaResponse::parse_metadata(void **buf, size_t *size) if (this->api_version >= 1) CHECK_RET(parse_i32(buf, size, &controller_id)); + KafkaMetaList *p_meta_list; + if (this->alien) + p_meta_list = &this->alien_meta_list; + else + p_meta_list = &this->meta_list; + CHECK_RET(kafka_meta_parse_topic(buf, size, this->api_version, - &this->meta_list, &this->broker_list)); + p_meta_list, &this->broker_list)); return 0; } @@ -3382,6 +3395,7 @@ static bool kafka_meta_find_topic(const std::string& topic_name, static int kafka_cgroup_parse_member(void **buf, size_t *size, KafkaCgroup *cgroup, KafkaMetaList *meta_list, + KafkaMetaList *alien_meta_list, int api_version) { int member_cnt = 0; @@ -3397,7 +3411,7 @@ static int kafka_cgroup_parse_member(void **buf, size_t *size, return -1; kafka_member_t **member = cgroup->get_members(); - std::set missing_topic; + std::set alien_topic; int i; for (i = 0; i < member_cnt; ++i) @@ -3443,7 +3457,7 @@ static int kafka_cgroup_parse_member(void **buf, size_t *size, list_add_tail(toppar->get_list(), &member[i]->toppar_list); if (!kafka_meta_find_topic(topic_name, meta_list)) - missing_topic.insert(topic_name); + alien_topic.insert(topic_name); } if (j != topic_cnt) @@ -3453,7 +3467,7 @@ static int kafka_cgroup_parse_member(void **buf, size_t *size, if (i != member_cnt) return -1; - for (const auto& v : missing_topic) + for (const auto& v : alien_topic) { KafkaMeta *meta = new KafkaMeta; if (!meta->set_topic(v)) @@ -3462,10 +3476,10 @@ static int kafka_cgroup_parse_member(void **buf, size_t *size, return -1; } - meta_list->add_item(std::move(*meta)); + alien_meta_list->add_item(std::move(*meta)); } - if (!missing_topic.empty()) + if (!alien_topic.empty()) cgroup->set_error(KAFKA_MISSING_TOPIC); return 0; @@ -3486,11 +3500,14 @@ int KafkaResponse::parse_joingroup(void **buf, size_t *size) CHECK_RET(parse_string(buf, size, &cgroup->leader_id)); CHECK_RET(parse_string(buf, size, &cgroup->member_id)); CHECK_RET(kafka_cgroup_parse_member(buf, size, &this->cgroup, - &this->meta_list, this->api_version)); + &this->meta_list, + &this->alien_meta_list, + this->api_version)); - if (this->cgroup.is_leader()) + if (cgroup->error != KAFKA_MISSING_TOPIC && this->cgroup.is_leader()) { CHECK_RET(this->cgroup.run_assignor(&this->meta_list, + &this->alien_meta_list, cgroup->protocol_name)); } diff --git a/src/protocol/KafkaMessage.h b/src/protocol/KafkaMessage.h index 1ddb797c..b12aae3d 100644 --- a/src/protocol/KafkaMessage.h +++ b/src/protocol/KafkaMessage.h @@ -93,6 +93,16 @@ public: return &this->meta_list; } + KafkaMetaList *get_alien_meta_list() + { + return &this->alien_meta_list; + } + + void set_alien() + { + this->alien = true; + } + void set_toppar_list(const KafkaTopparList& toppar_list) { this->toppar_list = toppar_list; @@ -127,10 +137,12 @@ public: this->cgroup = msg.cgroup; this->broker = msg.broker; this->meta_list = msg.meta_list; + this->alien_meta_list = msg.alien_meta_list; this->broker_list = msg.broker_list; this->toppar_list = msg.toppar_list; this->sasl = msg.sasl; this->api = msg.api; + this->alien = msg.alien; } void clear_buf() @@ -189,6 +201,7 @@ protected: KafkaCgroup cgroup; KafkaBroker broker; KafkaMetaList meta_list; + KafkaMetaList alien_meta_list; KafkaBrokerList broker_list; KafkaTopparList toppar_list; KafkaBuffer serialized; @@ -205,6 +218,8 @@ protected: kafka_sasl_t *sasl; kafka_api_t *api; + + bool alien; }; class KafkaRequest : public KafkaMessage diff --git a/tutorial/CMakeLists.txt b/tutorial/CMakeLists.txt index b648166a..830319da 100644 --- a/tutorial/CMakeLists.txt +++ b/tutorial/CMakeLists.txt @@ -76,7 +76,7 @@ endif() if (KAFKA STREQUAL "y") add_executable("kafka_cli" "tutorial-13-kafka_cli.cc") - target_link_libraries("kafka_cli" wfkafka workflow z snappy lz4 zstd rt) + target_link_libraries("kafka_cli" wfkafka ${WORKFLOW_LIB} z snappy lz4 zstd) endif () set(DIR10 tutorial-10-user_defined_protocol)