{"action":"created","issue":{"url":"https://api.github.com/repos/dataArtisans/cascading-flink/issues/60","repository_url":"https://api.github.com/repos/dataArtisans/cascading-flink","labels_url":"https://api.github.com/repos/dataArtisans/cascading-flink/issues/60/labels{/name}","comments_url":"https://api.github.com/repos/dataArtisans/cascading-flink/issues/60/comments","events_url":"https://api.github.com/repos/dataArtisans/cascading-flink/issues/60/events","html_url":"https://github.com/dataArtisans/cascading-flink/issues/60","id":153368501,"number":60,"title":"Use ConcurrentHashMap for Accumulators","user":{"login":"kkrugler","id":66217,"avatar_url":"https://avatars.githubusercontent.com/u/66217?v=3","gravatar_id":"","url":"https://api.github.com/users/kkrugler","html_url":"https://github.com/kkrugler","followers_url":"https://api.github.com/users/kkrugler/followers","following_url":"https://api.github.com/users/kkrugler/following{/other_user}","gists_url":"https://api.github.com/users/kkrugler/gists{/gist_id}","starred_url":"https://api.github.com/users/kkrugler/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/kkrugler/subscriptions","organizations_url":"https://api.github.com/users/kkrugler/orgs","repos_url":"https://api.github.com/users/kkrugler/repos","events_url":"https://api.github.com/users/kkrugler/events{/privacy}","received_events_url":"https://api.github.com/users/kkrugler/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"milestone":null,"comments":1,"created_at":"2016-05-06T02:53:58Z","updated_at":"2016-05-06T10:19:42Z","closed_at":null,"body":"While doing a quick \"poor man's profiler\" session with one of the TaskManager processes, I noticed that many (most?) of the threads that were actually running were in this state:\r\n\r\n```\r\n\"DataSource (/working1/terms) (8/20)\" daemon prio=10 tid=0x00007f55673e0800 nid=0x666a runnable [0x00007f556abcf000]\r\n   java.lang.Thread.State: RUNNABLE\r\n\tat java.util.Collections$SynchronizedMap.get(Collections.java:2037)\r\n\t- locked <0x00000006e73fe718> (a java.util.Collections$SynchronizedMap)\r\n\tat org.apache.flink.api.common.functions.util.AbstractRuntimeUDFContext.getAccumulator(AbstractRuntimeUDFContext.java:162)\r\n\tat org.apache.flink.api.common.functions.util.AbstractRuntimeUDFContext.getLongCounter(AbstractRuntimeUDFContext.java:113)\r\n\tat com.dataartisans.flink.cascading.runtime.util.FlinkFlowProcess.getOrInitCounter(FlinkFlowProcess.java:245)\r\n\tat com.dataartisans.flink.cascading.runtime.util.FlinkFlowProcess.increment(FlinkFlowProcess.java:128)\r\n\tat com.dataartisans.flink.cascading.runtime.util.FlinkFlowProcess.increment(FlinkFlowProcess.java:122)\r\n\tat cascading.tap.hadoop.util.MeasuredRecordReader.next(MeasuredRecordReader.java:65)\r\n\tat cascading.scheme.hadoop.SequenceFile.source(SequenceFile.java:97)\r\n\tat cascading.tuple.TupleEntrySchemeIterator.getNext(TupleEntrySchemeIterator.java:166)\r\n\tat cascading.tuple.TupleEntrySchemeIterator.hasNext(TupleEntrySchemeIterator.java:139)\r\n\tat com.dataartisans.flink.cascading.runtime.source.TapSourceStage.readNextRecord(TapSourceStage.java:70)\r\n\tat com.dataartisans.flink.cascading.runtime.source.TapInputFormat.reachedEnd(TapInputFormat.java:175)\r\n\tat org.apache.flink.runtime.operators.DataSourceTask.invoke(DataSourceTask.java:173)\r\n\tat org.apache.flink.runtime.taskmanager.Task.run(Task.java:559)\r\n\tat java.lang.Thread.run(Thread.java:745)\r\n```\r\n\r\nIt looks like Cascading is asking Flink to increment a counter with each Tuple read, and that in turn is often blocked on getting access to the ```Accumulator``` object in a map. It looks like this is a SynchronizedMap, but using a ConcurrentHashMap (for example) would reduce this contention."},"comment":{"url":"https://api.github.com/repos/dataArtisans/cascading-flink/issues/comments/217404935","html_url":"https://github.com/dataArtisans/cascading-flink/issues/60#issuecomment-217404935","issue_url":"https://api.github.com/repos/dataArtisans/cascading-flink/issues/60","id":217404935,"user":{"login":"fhueske","id":2388347,"avatar_url":"https://avatars.githubusercontent.com/u/2388347?v=3","gravatar_id":"","url":"https://api.github.com/users/fhueske","html_url":"https://github.com/fhueske","followers_url":"https://api.github.com/users/fhueske/followers","following_url":"https://api.github.com/users/fhueske/following{/other_user}","gists_url":"https://api.github.com/users/fhueske/gists{/gist_id}","starred_url":"https://api.github.com/users/fhueske/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/fhueske/subscriptions","organizations_url":"https://api.github.com/users/fhueske/orgs","repos_url":"https://api.github.com/users/fhueske/repos","events_url":"https://api.github.com/users/fhueske/events{/privacy}","received_events_url":"https://api.github.com/users/fhueske/received_events","type":"User","site_admin":false},"created_at":"2016-05-06T10:19:42Z","updated_at":"2016-05-06T10:19:42Z","body":"Hi Ken,\r\nthanks for profiling! We use Flink's Accumulators for collecting the counts and the Map is part of Flink's code base. So this would need to be fixed in Flink.\r\nCan you bring this issue to the Flink dev list or open a JIRA?\r\nThanks, Fabian"}}