Merge pull request #86 from sogou/master

merge from sogou/workflow

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}

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

src/CMakeLists.txt

@@ -55,6 +55,10 @@ add_library(
 	$<TARGET_OBJECTS:client>
 )
 
+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
 		$<TARGET_OBJECTS:client_kafka>
@@ -72,15 +75,18 @@ if (KAFKA STREQUAL "y")
 		$<TARGET_OBJECTS:protocol_kafka>
 		$<TARGET_OBJECTS:factory_kafka>
 	)
-	add_library(
-		${KAFKA_SHARED_LIB_NAME} SHARED
-		$<TARGET_OBJECTS:client_kafka>
-		$<TARGET_OBJECTS:util_kafka>
-		$<TARGET_OBJECTS:protocol_kafka>
-		$<TARGET_OBJECTS:factory_kafka>
-	)
 	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
+			$<TARGET_OBJECTS:client_kafka>
+			$<TARGET_OBJECTS:util_kafka>
+			$<TARGET_OBJECTS:protocol_kafka>
+			$<TARGET_OBJECTS:factory_kafka>
+		)
+		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 ()
 

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<std::string, MetaStatus>;
+		this->meta_map = new std::map<std::string, enum MetaStatus>;
 	}
 
 	~KafkaMember()
@@ -149,7 +149,7 @@ public:
 	KafkaBrokerList *broker_list;
 	KafkaBrokerMap *broker_map;
 	KafkaLockStatus *lock_status;
-	std::map<std::string, MetaStatus> *meta_map;
+	std::map<std::string, enum MetaStatus> *meta_map;
 
 private:
 	std::atomic<int> *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<int, int> *state_error;
-
+	bool flag = false;
 	for (size_t i = 0; i < pwork->size(); i++)
 	{
 		state_error = (std::pair<int, int> *)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;

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;

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;

src/protocol/HttpMessage.h

@@ -19,7 +19,6 @@
 #ifndef _HTTPMESSAGE_H_
 #define _HTTPMESSAGE_H_
 
-#include <sys/uio.h>
 #include <string.h>
 #include <utility>
 #include <string>
@@ -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

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<KafkaMetaSubscriber> *subscribers)
 {
-	std::vector<KafkaMetaSubscriber> 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<KafkaMetaSubscriber> 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);

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<kafka_member_t *> *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<kafka_member_t *> 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<KafkaMetaSubscriber> *subscribers);
 
 	static int kafka_range_assignor(kafka_member_t **members,
 									int member_elements,
@@ -1466,44 +1508,6 @@ private:
 	friend class KafkaList<KafkaBlock>;
 };
 
-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<kafka_member_t *> *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<kafka_member_t *> member_vec;
-};
-
 class KafkaBuffer
 {
 public:

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<std::string> missing_topic;
+	std::set<std::string> 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));
 	}
 

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

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)