{"action":"opened","issue":{"url":"https://api.github.com/repos/dailyburn/ratchet/issues/37","repository_url":"https://api.github.com/repos/dailyburn/ratchet","labels_url":"https://api.github.com/repos/dailyburn/ratchet/issues/37/labels{/name}","comments_url":"https://api.github.com/repos/dailyburn/ratchet/issues/37/comments","events_url":"https://api.github.com/repos/dailyburn/ratchet/issues/37/events","html_url":"https://github.com/dailyburn/ratchet/issues/37","id":223517606,"number":37,"title":"concurrent processing issue","user":{"login":"orelant","id":26119897,"avatar_url":"https://avatars2.githubusercontent.com/u/26119897?v=3","gravatar_id":"","url":"https://api.github.com/users/orelant","html_url":"https://github.com/orelant","followers_url":"https://api.github.com/users/orelant/followers","following_url":"https://api.github.com/users/orelant/following{/other_user}","gists_url":"https://api.github.com/users/orelant/gists{/gist_id}","starred_url":"https://api.github.com/users/orelant/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/orelant/subscriptions","organizations_url":"https://api.github.com/users/orelant/orgs","repos_url":"https://api.github.com/users/orelant/repos","events_url":"https://api.github.com/users/orelant/events{/privacy}","received_events_url":"https://api.github.com/users/orelant/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"assignees":[],"milestone":null,"comments":0,"created_at":"2017-04-21T22:55:35Z","updated_at":"2017-04-21T22:55:35Z","closed_at":null,"body":"Hi there,\r\n\r\nIm trying to run FuncTransformer concurrently and can't get it working.\r\nCould someone point me out what I am missing, please.\r\n\r\n```go\r\n\tlogger.LogLevel = logger.LevelDebug\r\n\r\n\tdb1, db2 := database.SetupDatabases()\r\n\tdb2.Query(\"delete from users\")\r\n\r\n\tread := processors.NewSQLReader(db1, \"SELECT id, name FROM users\")\r\n\tread.BatchSize = 1\r\n\ttransform := processors.NewFuncTransformer(func(d data.JSON) data.JSON {\r\n\t\tfmt.Println(string(d))\r\n\t\tjson := data.JSON(string(d))\r\n\t\ttime.Sleep(3 * time.Second)\r\n\t\treturn json\r\n\t})\r\n\ttransform.ConcurrencyLevel = 2\r\n\r\n\twrite := processors.NewSQLWriter(db2, \"users\")\r\n\r\n\tlayout, err := ratchet.NewPipelineLayout(\r\n\t\tratchet.NewPipelineStage(\r\n\t\t\tratchet.Do(read).Outputs(transform),\r\n\t\t),\r\n\t\tratchet.NewPipelineStage(\r\n\t\t\tratchet.Do(transform).Outputs(write),\r\n\t\t),\r\n\t\tratchet.NewPipelineStage(\r\n\t\t\tratchet.Do(write),\r\n\t\t),\r\n\t)\r\n\tif err != nil {\r\n\t\t// layout is invalid\r\n\t\tpanic(err.Error())\r\n\t}\r\n\r\n\t// Finally, create and run the Pipeline\r\n\tpipeline := ratchet.NewBranchingPipeline(layout)\r\n\terr = <-pipeline.Run()\r\n\r\n```\r\n\r\nI expect that code above will be executed within 3 secs but it takes 6 seconds.\r\nThere is a debug output:\r\n\r\n```\r\n2017/04/21 15:48:59 [Pipeline : connecting stages]\r\n2017/04/21 15:48:59 [Pipeline : sending GO to SQLReader]\r\n2017/04/21 15:48:59 [Pipeline - stage 3 SQLWriter waiting to receive data]\r\n2017/04/21 15:48:59 [Pipeline - stage 1 SQLReader waiting to receive data]\r\n2017/04/21 15:48:59 [Pipeline - stage 1 SQLReader received data]\r\n2017/04/21 15:48:59 [dataProcessor: processData SQLReader with concurrency = 0]\r\n2017/04/21 15:48:59 [SQLReader: Running -  SELECT id, name FROM users]\r\n2017/04/21 15:48:59 [Pipeline - stage 2 FuncTransformer waiting to receive data]\r\n2017/04/21 15:48:59 [Pipeline - stage 1 SQLReader input closed, calling Finish]\r\n2017/04/21 15:48:59 [Pipeline - stage 1 SQLReader closing output]\r\n2017/04/21 15:48:59 [Pipeline - stage 2 FuncTransformer received data]\r\n2017/04/21 15:48:59 [dataProcessor: processData FuncTransformer with concurrency = 2]\r\n2017/04/21 15:48:59 [dataProcessor: processData FuncTransformer waiting for work]\r\n2017/04/21 15:48:59 [dataProcessor: processData FuncTransformer work obtained]\r\n[{\"id\":1,\"name\":\"Aaron\"}]\r\n2017/04/21 15:48:59 [dataProcessor: processData FuncTransformer waiting to receive data on result chan]\r\n2017/04/21 15:49:02 [dataProcessor: processData FuncTransformer received data on result chan]\r\n2017/04/21 15:49:02 [dataProcessor: processData FuncTransformer done, releasing work]\r\n2017/04/21 15:49:02 [dataProcessor: sendResults checking for valid data to send]\r\n2017/04/21 15:49:02 [dataHandler: sendResults sending data]\r\n2017/04/21 15:49:02 [Pipeline - stage 2 FuncTransformer received data]\r\n2017/04/21 15:49:02 [dataProcessor: processData FuncTransformer with concurrency = 2]\r\n2017/04/21 15:49:02 [dataProcessor: processData FuncTransformer waiting for work]\r\n2017/04/21 15:49:02 [dataProcessor: processData FuncTransformer work obtained]\r\n[{\"id\":2,\"name\":\"Bob\"}]\r\n2017/04/21 15:49:02 [Pipeline - stage 3 SQLWriter received data]\r\n2017/04/21 15:49:02 [dataProcessor: processData SQLWriter with concurrency = 0]\r\n2017/04/21 15:49:02 [SQLWriter: Writing data...]\r\n2017/04/21 15:49:02 [SQLWriter: normal data scenario]\r\n2017/04/21 15:49:02 [dataProcessor: processData FuncTransformer waiting to receive data on result chan]\r\n2017/04/21 15:49:02 [SQLInsertData: building INSERT for len(objects) = 1]\r\n2017/04/21 15:49:02 [SQLInsertData: INSERT INTO users(id,name) VALUES(?,?) ON DUPLICATE KEY UPDATE `id`=VALUES(`id`),`name`=VALUES(`name`)]\r\n2017/04/21 15:49:02 [SQLInsertData: values [1 Aaron]]\r\n2017/04/21 15:49:02 [SQLInsertData: rows affected = 1, last insert ID = 0]\r\n2017/04/21 15:49:02 [SQLWriter: Write complete]\r\n2017/04/21 15:49:05 [dataProcessor: processData FuncTransformer received data on result chan]\r\n2017/04/21 15:49:05 [dataProcessor: processData FuncTransformer done, releasing work]\r\n2017/04/21 15:49:05 [dataProcessor: sendResults checking for valid data to send]\r\n2017/04/21 15:49:05 [dataHandler: sendResults sending data]\r\n2017/04/21 15:49:05 [Pipeline - stage 2 FuncTransformer input closed, calling Finish]\r\n2017/04/21 15:49:05 [Pipeline - stage 2 FuncTransformer closing output]\r\n2017/04/21 15:49:05 [Pipeline - stage 3 SQLWriter received data]\r\n2017/04/21 15:49:05 [dataProcessor: processData SQLWriter with concurrency = 0]\r\n2017/04/21 15:49:05 [SQLWriter: Writing data...]\r\n2017/04/21 15:49:05 [SQLWriter: normal data scenario]\r\n2017/04/21 15:49:05 [SQLInsertData: building INSERT for len(objects) = 1]\r\n2017/04/21 15:49:05 [SQLInsertData: INSERT INTO users(id,name) VALUES(?,?) ON DUPLICATE KEY UPDATE `id`=VALUES(`id`),`name`=VALUES(`name`)]\r\n2017/04/21 15:49:05 [SQLInsertData: values [2 Bob]]\r\n2017/04/21 15:49:05 [SQLInsertData: rows affected = 1, last insert ID = 0]\r\n2017/04/21 15:49:05 [SQLWriter: Write complete]\r\n2017/04/21 15:49:05 [Pipeline - stage 3 SQLWriter input closed, calling Finish]\r\n2017/04/21 15:49:05 [Pipeline - stage 3 SQLWriter closing output]\r\n```\r\n\r\nThank you in advance."}}
{"actor":{"display_login":"orelant"}}