Anyone have a plain PG example, i.e. not using ActiveRecord or other ORM?

[jreidthompson]

I can't seem to find a way to make it work for threads. I think I have a process version working.

[grosser]

If you have the process version working post some code, I saw weird issues in the tests we did so far #83

[jreidthompson]

On Tue, 2013-10-15 at 21:40 -0700, Michael Grosser wrote:

If you have the process version working post some code, I saw weird
issues in the tests we did so far #83

—
Reply to this email directly or view it on GitHub.

$ cat testProcesses.rb
require 'rubygems'
require 'pg'
require 'pp'
require 'parallel'

dbname='test'
dbuser='username'
dbpwd=''
pghost='localhost'
#pgport='6543' # pgbouncer pool connection port
pgport='5432'

#######################################################################

$ pg_dump -s -t test test

--

-- PostgreSQL database dump

--

SET statement_timeout = 0;

SET client_encoding = 'UTF8';

SET standard_conforming_strings = on;

SET check_function_bodies = false;

SET client_min_messages = warning;

SET search_path = public, pg_catalog;

SET default_tablespace = '';

SET default_with_oids = false;

--

-- Name: test; Type: TABLE; Schema: public; Owner: rthompso; Tablespace:

--

CREATE TABLE test (

a integer,

str text,

myserial integer NOT NULL

);

ALTER TABLE public.test OWNER TO rthompso;

--

-- Name: test_myserial_seq; Type: SEQUENCE; Schema: public; Owner: rthompso

--

CREATE SEQUENCE test_myserial_seq

START WITH 1

INCREMENT BY 1

NO MINVALUE

NO MAXVALUE

CACHE 1;

ALTER TABLE public.test_myserial_seq OWNER TO rthompso;

--

-- Name: test_myserial_seq; Type: SEQUENCE OWNED BY; Schema: public; Owner: rthompso

--

ALTER SEQUENCE test_myserial_seq OWNED BY test.myserial;

--

-- Name: myserial; Type: DEFAULT; Schema: public; Owner: rthompso

--

ALTER TABLE ONLY test ALTER COLUMN myserial SET DEFAULT nextval('test_myserial_seq'::regclass);

--

-- PostgreSQL database dump complete

--

#######################################################################

#######################################################################

$ cat /tmp/testdata

1 '6652491'

2 '6578035'

3 '6607464'

4 '6610893'

5 '6620404'

... snip ...

1700 '7003391'

#######################################################################

File.foreach("/tmp/testdata").each_slice(60) do |line_array|
# line_array is an array containing N lines.
# at this point you can iterate over these lines
# and spawn off your processes/threads
#
Parallel.each(line_array, :in_processes => 15) {|line|

    puts "process line: #{line}"

    @DB1Connection = PGconn.connect(:dbname => dbname, :user => dbuser, :password => dbpwd, :host => pghost, :port => pgport)
    if (@DB1Connection.nil?)
        puts "Unable to connect to database ateb"
        exit 1
    end

    res = @DB1Connection.prepare("insert_post_response", "INSERT into test(a, str) values($1::integer, $2::text);")
    if ( res.result_status > 4 )
        puts "DB prepare error: " + res.result_status.to_s
        res.clear
        exit 1
    end
    res.clear

    count, value = line.split
    res = @DB1Connection.exec_prepared('insert_post_response', [count, value])
    if ( res.result_status > 4 )
        puts "DB insert error: " + res.result_status.to_s
    end
    res.clear

    puts "end processing  #{$$.to_s}"

you could prob leave off this close, as the process exit will close the connection

    res = @DB1Connection.close
    if ( res.result_status > 4 )
        @log.error "DB insert error: " + res.result_status.to_s
    end
    res.clear
}

end

[grosser]

you should be able to do @connection ||= PGconn.connect as every fork only
needs 1 connection (forks work on multiple items serially)

On Wed, Oct 16, 2013 at 11:42 AM, jreidthompson notifications@github.comwrote:

On Tue, 2013-10-15 at 21:40 -0700, Michael Grosser wrote:

If you have the process version working post some code, I saw weird
issues in the tests we did so far #83

—
Reply to this email directly or view it on GitHub.

$ cat testProcesses.rb
require 'rubygems'
require 'pg'
require 'pp'
require 'parallel'

dbname='test'
dbuser='username'
dbpwd=''
pghost='localhost'
#pgport='6543' # pgbouncer pool connection port
pgport='5432'

#######################################################################

$ pg_dump -s -t test test

--

-- PostgreSQL database dump

--

SET statement_timeout = 0;

SET client_encoding = 'UTF8';

SET standard_conforming_strings = on;

SET check_function_bodies = false;

SET client_min_messages = warning;

SET search_path = public, pg_catalog;

SET default_tablespace = '';

SET default_with_oids = false;

--

-- Name: test; Type: TABLE; Schema: public; Owner: rthompso; Tablespace:

--

CREATE TABLE test (

a integer,

str text,

myserial integer NOT NULL

);

ALTER TABLE public.test OWNER TO rthompso;

--

-- Name: test_myserial_seq; Type: SEQUENCE; Schema: public; Owner:

rthompso

--

CREATE SEQUENCE test_myserial_seq

START WITH 1

INCREMENT BY 1

NO MINVALUE

NO MAXVALUE

CACHE 1;

ALTER TABLE public.test_myserial_seq OWNER TO rthompso;

--

-- Name: test_myserial_seq; Type: SEQUENCE OWNED BY; Schema: public;

Owner: rthompso

--

ALTER SEQUENCE test_myserial_seq OWNED BY test.myserial;

--

-- Name: myserial; Type: DEFAULT; Schema: public; Owner: rthompso

--

ALTER TABLE ONLY test ALTER COLUMN myserial SET DEFAULT

nextval('test_myserial_seq'::regclass);

--

-- PostgreSQL database dump complete

--

#######################################################################

#######################################################################

$ cat /tmp/testdata

1 '6652491'

2 '6578035'

3 '6607464'

4 '6610893'

5 '6620404'

... snip ...

1700 '7003391'

#######################################################################

File.foreach("/tmp/testdata").each_slice(60) do |line_array|

line_array is an array containing N lines.

at this point you can iterate over these lines

and spawn off your processes/threads

Parallel.each(line_array, :in_processes => 15) {|line|

puts "process line: #{line}"

@DB1Connection = PGconn.connect(:dbname => dbname, :user => dbuser,
:password => dbpwd, :host => pghost, :port => pgport)
if (@DB1Connection.nil?)
puts "Unable to connect to database ateb"
exit 1
end

res = @DB1Connection.prepare("insert_post_response", "INSERT into test(a,
str) values($1::integer, $2::text);")
if ( res.result_status > 4 )
puts "DB prepare error: " + res.result_status.to_s
res.clear
exit 1
end
res.clear

count, value = line.split
res = @DB1Connection.exec_prepared('insert_post_response', [count, value])
if ( res.result_status > 4 )
puts "DB insert error: " + res.result_status.to_s
end
res.clear

puts "end processing #{$$.to_s}"

you could prob leave off this close, as the process exit will close the

connection
res = @DB1Connection.close
if ( res.result_status > 4 )
@log.error "DB insert error: " + res.result_status.to_s
end
res.clear
}
end

—
Reply to this email directly or view it on GitHubhttps://github.com//issues/84#issuecomment-26445579
.

[jreidthompson]

On Tue, 2013-10-15 at 21:40 -0700, Michael Grosser wrote:

If you have the process version working post some code, I saw weird issues in the tests we did so far #83

—
Reply to this email directly or view it on GitHub.

this appears to work -- threading... but would need validation...

$ ruby -v
ruby 1.8.7 (2013-06-27 patchlevel 374) [x86_64-linux]

require 'rubygems'
require 'pg'
require 'pp'
require 'parallel'

dbname='test'
dbuser='username'
dbpwd=''
pghost='localhost'
#pgport='6543'
pgport='5432'

File.foreach("/tmp/testdata").each_slice(100) do |line_array|
# line_array is an array containing N lines.
# at this point you can iterate over these lines
# and spawn off your processes/threads
#

$DB1Connection = PGconn.connect(:dbname => dbname, :user => dbuser, :password => dbpwd, :host => pghost, :port => pgport)
if ($DB1Connection.nil?)
    puts "Unable to connect to database ateb"
    exit 1
end

res = $DB1Connection.prepare("insert_post_response", "INSERT into test(a, str) values($1::integer, $2::text);")
if ( res.result_status > 4 )
    puts "DB prepare error: " + res.result_status.to_s
    res.clear
    exit 1
end

Parallel.each(line_array, :in_threads => 20) {|line|
    puts "process line: #{line}"


    count, value = line.split
    res = $DB1Connection.exec_prepared('insert_post_response', [count, value])
    if ( res.result_status > 4 )
        puts "DB insert error: " #+ res.result_status.to_s
    end


    puts "end processing  #{id.to_s}"

    #        res = $DB1Connection.close
    #        if ( res.result_status > 4 )
    #            @log.error "DB insert error: " + res.result_status.to_s
    #        end
    #        res.clear

}
$DB1Connection.close

end

[jreidthompson]

On Wed, 2013-10-16 at 11:47 -0700, Michael Grosser wrote:

you should be able to do @connection ||= PGconn.connect as every fork
only
needs 1 connection (forks work on multiple items serially)

slightly confused

1. i'm not familiar with the convention "@connection ||= PGconn.connect"
   is that just changing
   @DB1Connection = PGconn.connect(:dbname => dbname, :user => dbuser, :password => dbpwd, :host => pghost, :port => pgport)
   to
   @DB1Connection ||= PGconn.connect(:dbname => dbname, :user => dbuser, :password => dbpwd, :host => pghost, :port => pgport)

   and does it mean,
    use @DB1Connection if it exists else create and assign
    connection to @DB1Connection

2. I thought that what I implemented was doing just that -- creating a
   single DB connection for each process which processed 60 lines then
   exited

3. I don't really understand how to configure 1)
   just changing the code per 1) doesn't work... the script errors out
   after the 15th line..
   $ ruby testProcesses.rb
   I, [2013-10-16 15:51:16#6117] INFO -- : process line: 1 '6652491'

I, [2013-10-16 15:51:16#6118] INFO -- : process line: 2 '6578035'

I, [2013-10-16 15:51:16#6119] INFO -- : process line: 3 '6607464'

I, [2013-10-16 15:51:16#6120] INFO -- : process line: 4 '6610893'

I, [2013-10-16 15:51:16#6121] INFO -- : process line: 5 '6620404'

I, [2013-10-16 15:51:16#6122] INFO -- : process line: 6 '6626819'

I, [2013-10-16 15:51:16#6123] INFO -- : process line: 7 '6584929'

I, [2013-10-16 15:51:16#6124] INFO -- : process line: 8 '6320761'

I, [2013-10-16 15:51:16#6125] INFO -- : process line: 9 '6320985'

I, [2013-10-16 15:51:16#6126] INFO -- : process line: 10 '7001250'

I, [2013-10-16 15:51:16#6127] INFO -- : process line: 11 '6587709'

I, [2013-10-16 15:51:16#6128] INFO -- : process line: 12 '6611125'

I, [2013-10-16 15:51:16#6129] INFO -- : process line: 13 '6630632'

I, [2013-10-16 15:51:16#6131] INFO -- : process line: 15 '6657703'
I, [2013-10-16 15:51:16#6130] INFO -- : process line: 14 '6648861'

I, [2013-10-16 15:51:16#6118] INFO -- : end processing 6118
I, [2013-10-16 15:51:16#6118] INFO -- : process line: 16 '6661178'

I, [2013-10-16 15:51:16#6122] INFO -- : end processing 6122
I, [2013-10-16 15:51:16#6129] INFO -- : end processing 6129
I, [2013-10-16 15:51:16#6124] INFO -- : end processing 6124
I, [2013-10-16 15:51:16#6126] INFO -- : end processing 6126
I, [2013-10-16 15:51:16#6130] INFO -- : end processing 6130
I, [2013-10-16 15:51:16#6131] INFO -- : end processing 6131
I, [2013-10-16 15:51:16#6120] INFO -- : end processing 6120
I, [2013-10-16 15:51:16#6125] INFO -- : end processing 6125
I, [2013-10-16 15:51:16#6123] INFO -- : end processing 6123
I, [2013-10-16 15:51:16#6127] INFO -- : end processing 6127
I, [2013-10-16 15:51:16#6121] INFO -- : end processing 6121
I, [2013-10-16 15:51:16#6117] INFO -- : end processing 6117
I, [2013-10-16 15:51:16#6128] INFO -- : end processing 6128
I, [2013-10-16 15:51:16#6119] INFO -- : end processing 6119
/usr/local/lib64/ruby/gems/1.8/gems/parallel-0.6.1/lib/parallel.rb:224:in `work_in_processes': Undumpable Exception -- #<PG::Error: ERROR: prepared statement "insert_post_response" already exists (RuntimeError)

from /usr/local/lib64/ruby/gems/1.8/gems/parallel-0.6.1/lib/parallel.rb:103:in `map'
from /usr/local/lib64/ruby/gems/1.8/gems/parallel-0.6.1/lib/parallel.rb:78:in `each'
from testProcesses.rb:21
from /usr/lib64/ruby/site_ruby/1.8/rubygems/core_ext/kernel_require.rb:110:in `each_slice'
from /usr/lib64/ruby/site_ruby/1.8/rubygems/core_ext/kernel_require.rb:110:in `each'
from testProcesses.rb:15:in `foreach'
from testProcesses.rb:15:in `each'
from testProcesses.rb:15:in `each_slice'
from testProcesses.rb:15

[grosser]

1: yes
2: no - it creates 1 connection in each fork, parallel only forks
:in_processes (15) times and then works on multiple items serially in each
fork to reduce fork overhead
3: No idea, maybe fixed with this change...

On Wed, Oct 16, 2013 at 12:51 PM, jreidthompson notifications@github.comwrote:

On Wed, 2013-10-16 at 11:47 -0700, Michael Grosser wrote:

you should be able to do @connection ||= PGconn.connect as every fork
only
needs 1 connection (forks work on multiple items serially)

slightly confused

1. i'm not familiar with the convention "@connection ||= PGconn.connect"
   is that just changing
   @DB1Connection = PGconn.connect(:dbname => dbname, :user => dbuser,
   :password => dbpwd, :host => pghost, :port => pgport)
   to
   @DB1Connection ||= PGconn.connect(:dbname => dbname, :user => dbuser,
   :password => dbpwd, :host => pghost, :port => pgport)

and does it mean,
use @DB1Connection if it exists else create and assign
connection to @DB1Connection

2. I thought that what I implemented was doing just that -- creating a
   single DB connection for each process which processed 60 lines then
   exited

3. I don't really understand how to configure 1)
   just changing the code per 1) doesn't work... the script errors out
   after the 15th line..
   $ ruby testProcesses.rb
   I, [2013-10-16 15:51:16#6117] INFO -- : process line: 1 '6652491'

I, [2013-10-16 15:51:16#6118] INFO -- : process line: 2 '6578035'

I, [2013-10-16 15:51:16#6119] INFO -- : process line: 3 '6607464'

I, [2013-10-16 15:51:16#6120] INFO -- : process line: 4 '6610893'

I, [2013-10-16 15:51:16#6121] INFO -- : process line: 5 '6620404'

I, [2013-10-16 15:51:16#6122] INFO -- : process line: 6 '6626819'

I, [2013-10-16 15:51:16#6123] INFO -- : process line: 7 '6584929'

I, [2013-10-16 15:51:16#6124] INFO -- : process line: 8 '6320761'

I, [2013-10-16 15:51:16#6125] INFO -- : process line: 9 '6320985'

I, [2013-10-16 15:51:16#6126] INFO -- : process line: 10 '7001250'

I, [2013-10-16 15:51:16#6127] INFO -- : process line: 11 '6587709'

I, [2013-10-16 15:51:16#6128] INFO -- : process line: 12 '6611125'

I, [2013-10-16 15:51:16#6129] INFO -- : process line: 13 '6630632'

I, [2013-10-16 15:51:16#6131] INFO -- : process line: 15 '6657703'
I, [2013-10-16 15:51:16#6130] INFO -- : process line: 14 '6648861'

I, [2013-10-16 15:51:16#6118] INFO -- : end processing 6118
I, [2013-10-16 15:51:16#6118] INFO -- : process line: 16 '6661178'

I, [2013-10-16 15:51:16#6122] INFO -- : end processing 6122
I, [2013-10-16 15:51:16#6129] INFO -- : end processing 6129
I, [2013-10-16 15:51:16#6124] INFO -- : end processing 6124
I, [2013-10-16 15:51:16#6126] INFO -- : end processing 6126
I, [2013-10-16 15:51:16#6130] INFO -- : end processing 6130
I, [2013-10-16 15:51:16#6131] INFO -- : end processing 6131
I, [2013-10-16 15:51:16#6120] INFO -- : end processing 6120
I, [2013-10-16 15:51:16#6125] INFO -- : end processing 6125
I, [2013-10-16 15:51:16#6123] INFO -- : end processing 6123
I, [2013-10-16 15:51:16#6127] INFO -- : end processing 6127
I, [2013-10-16 15:51:16#6121] INFO -- : end processing 6121
I, [2013-10-16 15:51:16#6117] INFO -- : end processing 6117
I, [2013-10-16 15:51:16#6128] INFO -- : end processing 6128
I, [2013-10-16 15:51:16#6119] INFO -- : end processing 6119
/usr/local/lib64/ruby/gems/1.8/gems/parallel-0.6.1/lib/parallel.rb:224:in
`work_in_processes': Undumpable Exception -- #<PG::Error: ERROR: prepared
statement "insert_post_response" already exists (RuntimeError)

from
/usr/local/lib64/ruby/gems/1.8/gems/parallel-0.6.1/lib/parallel.rb:103:in
map' from /usr/local/lib64/ruby/gems/1.8/gems/parallel-0.6.1/lib/parallel.rb:78:in each'
from testProcesses.rb:21
from
/usr/lib64/ruby/site_ruby/1.8/rubygems/core_ext/kernel_require.rb:110:in
each_slice' from /usr/lib64/ruby/site_ruby/1.8/rubygems/core_ext/kernel_require.rb:110:in each'
from testProcesses.rb:15:in foreach' from testProcesses.rb:15:ineach'
from testProcesses.rb:15:in `each_slice'
from testProcesses.rb:15

—
Reply to this email directly or view it on GitHubhttps://github.com//issues/84#issuecomment-26451535
.

[jreidthompson]

On 10/16/2013 3:56 PM, Michael Grosser wrote:

1: yes
2: no - it creates 1 connection in each fork, parallel only forks
:in_processes (15) times and then works on multiple items serially in
each
fork to reduce fork overhead
3: No idea, maybe fixed with this change...

ok I see
2000 line file --> /tmp/testdata

File.foreach("/tmp/testdata").each_slice(1000) do |line_array|
Parallel.each(line_array, :in_processes => 15) {|line|
# do work
end

generates a total of 30 forked processes.
one chunk of 1000 lines worked on by the first group of 15 forked
processes
1000/15 = ~66 lines per process
one chunk of 1000 lines worked on by the second group of 15 forked
processes
1000/15 = ~66 lines per process

so, modifying the previous test a little bit, then validation =

add pid column to table
$ psql -c "\d test " test
Table "public.test"
Column | Type | Modifiers
----------+---------+---------------------------------------------------------
a | integer |
str | text |
myserial | integer | not null default
nextval('test_myserial_seq'::regclass)
pid | integer |

alter the script to manage the dbconnection and prepared statement per
your earlier response

File.foreach("/tmp/testdata").each_slice(1000) do |line_array|
# line_array is an array containing N lines.
# at this point you can iterate over these lines
# and spawn off your processes/threads
#
Parallel.each(line_array, :in_processes => 15) {|line|

     @DB1Connection ||= PGconn.connect(:dbname => dbname, :user => 

dbuser, :password => dbpwd, :host => pghost, :port => pgport)
if (@DB1Connection.nil?)
puts "Unable to connect to database ateb"
exit 1
end

     @prep ||= @DB1Connection.prepare("insert_post_response", 

"INSERT into test(a, str, pid) values($1::integer, $2::text, $3::integer);")
if ( @prep.result_status > 4 )
puts "DB prepare error: " + @prep.result_status.to_s
@prep.clear
exit 1
end

     count, value = line.split
     res = @DB1Connection.exec_prepared('insert_post_response', 

[count, value, $$])
if ( res.result_status > 4 )
puts "DB insert error: " + res.result_status.to_s
end
res.clear

puts "end processing #{$$.to_s}"

have to leave this out, or it bails out after the first 15 inserts

you could prob leave off this close, as the process exit will close

the connection

res = @DB1Connection.close

if ( res.result_status > 4 )

@log.error "DB insert error: " + res.result_status.to_s

end

res.clear

 }

end

....
$ psql -c "truncate test" test
TRUNCATE TABLE
[19:21:00][0s] rthompso@raker2.ateb.com> /tmp
$ psql -c "select count(*), pid from test group by pid" test
count | pid
-------+-----
(0 rows)

[19:21:07][0s] rthompso@raker2.ateb.com> /tmp
$ wc -l testdata
2000 testdata
[19:21:16][0s] rthompso@raker2.ateb.com> /tmp
$ ruby testProcesses.rb
[19:21:42][15s] rthompso@raker2.ateb.com> /tmp
$ psql -c "select count(*) from test" test

count

2000
(1 row)

[19:21:48][0s] rthompso@raker2.ateb.com> /tmp
$ psql -c "select count(*), pid from test group by pid" test
count | pid
-------+-------
67 | 14922
66 | 14916
67 | 14954
67 | 14927
66 | 14959
67 | 14928
67 | 14958
67 | 14967
66 | 14957
67 | 14966
66 | 14961
67 | 14965
67 | 14955
67 | 14914
67 | 14964
67 | 14963
67 | 14926
67 | 14917
66 | 14919
67 | 14953
66 | 14920
66 | 14921
67 | 14915
67 | 14924
66 | 14960
66 | 14918
67 | 14923
67 | 14956
66 | 14962
67 | 14925
(30 rows)

[jreidthompson]

prob should mention version
parallel (0.6.1)

[jreidthompson]

didn't realize I was that far behind -- updated to latest
parallel (0.9.0)
process test appears to work the same

[grosser]

But it still fails right ? Not sure what to do :/

[jreidthompson]

same code as in process test except for:
#Parallel.each(line_array, :in_processes => 15) {|line|
Parallel.each(line_array, :in_threads => 15) {|line|
.. all inserts done by the one GIL thread
$ time ruby testProcesses.rb

real 0m38.840s
user 0m0.427s
sys 0m0.082s
[19:58:15][39s] rthompso@raker2.ateb.com> /tmp
$ psql -c "select count(*), pid from test group by pid" test
count | pid
-------+-------
2000 | 28224
(1 row)

------some quick simple tests ---

15 processes/2000 line file/1 2000 line chunk

[19:59:46][7s] rthompso@raker2.ateb.com> /tmp
TRUNCATE TABLE
[20:00:01][10s] rthompso@raker2.ateb.com> /tmp
$ time ruby testProcesses.rb

real 0m8.039s
user 0m0.520s
sys 0m0.194s
[20:00:21][8s] rthompso@raker2.ateb.com> /tmp
$ time ruby testProcesses.rb

real 0m14.124s
user 0m0.522s
sys 0m0.224s
[20:00:37][14s] rthompso@raker2.ateb.com> /tmp
$ time ruby testProcesses.rb

real 0m4.941s
user 0m0.538s
sys 0m0.183s
[20:00:44][5s] rthompso@raker2.ateb.com> /tmp
$ time ruby testProcesses.rb

real 0m4.934s
user 0m0.526s
sys 0m0.205s

[20:00:51][5s] rthompso@raker2.ateb.com> /tmp
$ vi testProcesses.rb

15 ?threads? (GIL) /2000 line file/1 2000 line chunk

[20:01:06][6s] rthompso@raker2.ateb.com> /tmp
$ psql -c "truncate test" test
TRUNCATE TABLE
[20:01:13][2s] rthompso@raker2.ateb.com> /tmp
$ time ruby testProcesses.rb

real 1m9.523s
user 0m0.451s
sys 0m0.104s
[20:02:26][70s] rthompso@raker2.ateb.com> /tmp
$ time ruby testProcesses.rb

real 1m21.338s
user 0m0.441s
sys 0m0.123s
[20:03:50][82s] rthompso@raker2.ateb.com> /tmp
$ time ruby testProcesses.rb

real 1m7.278s
user 0m0.434s
sys 0m0.106s

[jreidthompson]

no -- not failing -- these appear to be working for me