-- Parallel PL/SQL Launcher, © William Robertson 2007 www.williamrobertson.net
-- Experimental package for submitting PL/SQL in parallel using the Oracle SQL Parallel Query mechanism.
--
-- Many have complained that PL/SQL has no multi-threading support, as Java (for example) does. The conventional solution,
-- if one is really needed (arguably it is never a good idea anyway) is to use DBMS_JOB or DBMS_SCHEDULER, and if necessary
-- write your own wrapper to handle completion status reporting (Has it finished yet? Did it succeed?) - for an example of
-- this approach , see http://www.williamrobertson.net/code/job_ot.typ
--
-- However, SQL does have a parallel threading mechanism, in the form of Parallel Query (PQ). You simply execute your
-- query using a PARALLEL() hint, or against tables with a parallel degree defined - particularly effective with partitioned
-- tables, where each PQ slave can take one partition (depending on resources available) and combine results at the end.
-- You can think of this as a PQ controller dividing up the job, handing out tasks to PQ slaves, and collating the results.
-- Also, PL/SQL lets you define parallel-enabled pipelined functions that accept a ref cursor, fetch the rows in parallel,
-- and return the results as they come back from the PQ slaves.
--
-- Now, if you create a table "PQ_DRIVER" with 4 partitions and a parallel degree of 4, with one row in each partition, and
-- you execute a query along the lines of "SELECT /*+ PARALLEL(pq,4) */ thread_id FROM pq_driver pq", that should persuade the
-- PQ controller to set four PQ slave processes to work on it (one per partition). And if you pass that query as a cursor
-- parameter to a parallel-enabled pipelined function, then shouldn't that create a situation where each each row is
-- processed by a separate PQ slave process? So here is a way to use (alright, hack) the PQ engine so that it processes
-- arbitrary PL/SQL procedure calls in parallel.
--
-- This has not been stress tested and I make no promises that it will work. In particular I have found that the degree of
-- parallelism is different (lower) in 11g than in 10g - possibly 11g is more intelligent and decides that four slaves would
-- be excessive for a table with only four rows. Let me know what you find.

-- Update: 2008-08-01:
-- In my home 11.1.0.6.0 EE test database, I noticed initialisation parameters CPU_COUNT=1 (which was true, there is only 1 cpu)
-- and PARALLEL_THREADS_PER_CPU=2 (the default setting, and again reasonable). I found that setting PARALLEL_THREADS_PER_CPU = 4
-- (or CPU_COUNT = 2) enabled the full 4 parallel threads requested in parallel_launch.test. Also since gathering statistics on
-- the PQ_DRIVER table had the effect of serialising queries against it, and since by default the background stats job will do
-- this, I added a call to dbms_stats.lock_table_stats(user,'pq_driver').

-- Note also, the schema requires select privilege on v$mystat, v$px_session and v$session.


DROP TABLE pq_driver PURGE;
DROP TABLE log_times PURGE;

CREATE TYPE varchar2_tt AS TABLE OF VARCHAR2(4000)
/

GRANT EXECUTE ON varchar2_tt TO PUBLIC;

CREATE TABLE pq_driver
( thread_id NUMBER(1) NOT NULL PRIMARY KEY )
PARTITION BY LIST(thread_id)
( PARTITION p1 VALUES(1)
, PARTITION p2 VALUES(2)
, PARTITION p3 VALUES(3)
, PARTITION p4 VALUES(4) )
PARALLEL 4
/

INSERT ALL
INTO pq_driver VALUES (1)
INTO pq_driver VALUES (2)
INTO pq_driver VALUES (3)
INTO pq_driver VALUES (4)
SELECT * FROM dual;

COMMENT ON TABLE pq_driver IS 'Control table for generating parallel ref cursors with package parallel_launch';

CALL dbms_stats.lock_table_stats(user,'pq_driver');

-- PQ selection is sensitive to table stats. Analyzing table with
-- exec DBMS_STATS.GATHER_TABLE_STATS(user,'pq_driver')
-- causes serialisation.
-- Will automatic stats gathering have the same effect?

CREATE TABLE log_times
( thread_id   INTEGER NOT NULL CONSTRAINT log_times_pk PRIMARY KEY
, what        VARCHAR2(100)
, sid         INTEGER
, serial#     INTEGER
, start_time  TIMESTAMP DEFAULT SYSTIMESTAMP NOT NULL
, end_time    TIMESTAMP
, label       VARCHAR2(30)
, errors      VARCHAR2(1000) );

CREATE TRIGGER log_times_defaults_trg
BEFORE INSERT ON log_times
FOR EACH ROW
BEGIN
    SELECT sid, serial#
    INTO   :new.sid, :new.serial#
    FROM   v$session
    WHERE  sid IN
           ( SELECT sid FROM v$mystat );
END;
/

show errors



CREATE OR REPLACE PACKAGE parallel_launch
AS
	TYPE rc_pq_driver IS REF CURSOR RETURN pq_driver%ROWTYPE;

	-- PQ launch vehicle:
	-- Must be strongly typed ref cursor to allow partition by range or hash
	-- Must be in package spec to be visible to SQL
	-- Must be an autonomous transaction if we are doing any DML in the submitted procedures

	FUNCTION pq_submit
		( p_job_list  VARCHAR2_TT
		, p_pq_refcur rc_pq_driver )
		RETURN varchar2_tt
		PARALLEL_ENABLE(PARTITION p_pq_refcur BY ANY)
		PIPELINED;

	PROCEDURE submit
		( p_job1 VARCHAR2
		, p_job2 VARCHAR2
		, p_job3 VARCHAR2
		, p_job4 VARCHAR2 );

	-- Convenient test procedure - calls 'parallel_launch.slow_proc()' 4 times to see whether they all run at once:
	PROCEDURE test;

	-- Dummy procedure for testing - calls dbms_lock.sleep(2)
	PROCEDURE slow_proc( p_id INTEGER );
END parallel_launch;
/

show errors

CREATE OR REPLACE PACKAGE BODY parallel_launch
AS
	TYPE stats_rec IS RECORD
	( thread_id            log_times.thread_id%TYPE
	, what                 log_times.what%TYPE
	, start_timestr        VARCHAR2(8)
	, sid                  v$mystat.sid%TYPE
	, serial#              v$px_session.serial#%TYPE
	, pq_actual_degree     NUMBER
	, pq_requested_degree  NUMBER
	, rowcount             PLS_INTEGER := 0 );

	g_clear_stats_rec stats_rec;


	PROCEDURE log_start
		( p_thread_id  log_times.thread_id%TYPE
		, p_what       log_times.what%TYPE )
	IS
		PRAGMA AUTONOMOUS_TRANSACTION;
	BEGIN
		DELETE log_times WHERE thread_id = p_thread_id;

		INSERT INTO log_times
		( thread_id, what )
		VALUES
		( p_thread_id, p_what );

		COMMIT;
	END log_start;


	PROCEDURE log_end
		( p_thread_id  log_times.thread_id%TYPE
		, p_errors     log_times.errors%TYPE DEFAULT NULL )
	IS
		PRAGMA AUTONOMOUS_TRANSACTION;
	BEGIN
		UPDATE log_times
		SET    end_time = SYSTIMESTAMP
		     , errors = p_errors
		WHERE  thread_id = p_thread_id;

		COMMIT;
	END log_end;


	PROCEDURE execute_command
		( p_what log_times.what%TYPE )
	IS
		PRAGMA AUTONOMOUS_TRANSACTION;
	BEGIN
		EXECUTE IMMEDIATE p_what;
		COMMIT;
	END execute_command;


	FUNCTION pq_submit
		( p_job_list  VARCHAR2_TT
		, p_pq_refcur rc_pq_driver )
		RETURN varchar2_tt
		PARALLEL_ENABLE(PARTITION p_pq_refcur BY ANY)
		PIPELINED
	IS
		v_error_text VARCHAR2(2000);
		r pq_driver%ROWTYPE;
		r_row_stats stats_rec;
	BEGIN
		LOOP
			FETCH p_pq_refcur INTO r;
			EXIT WHEN p_pq_refcur%NOTFOUND;

			SELECT TO_CHAR(SYSDATE,'HH24:MI:SS')
			     , s.sid
			     , pqs.serial#
			     , pqs.degree
			     , pqs.req_degree
			INTO   r_row_stats.start_timestr
			     , r_row_stats.sid
			     , r_row_stats.serial#
			     , r_row_stats.pq_actual_degree
			     , r_row_stats.pq_requested_degree
			FROM   ( SELECT sid FROM v$mystat WHERE rownum = 1 ) s
			       LEFT JOIN v$px_session pqs ON pqs.sid = s.sid;

			r_row_stats.thread_id := r.thread_id;
			r_row_stats.rowcount := p_pq_refcur%ROWCOUNT;
			r_row_stats.what := 'BEGIN ' || RTRIM(p_job_list(r.thread_id),';') || '; END;';

			BEGIN
				log_start(r.thread_id, r_row_stats.what);

				execute_command(r_row_stats.what);

				log_end(r.thread_id);

				PIPE ROW
				( RPAD('sid=' || r_row_stats.sid || ' serial#=' || r_row_stats.serial# || ':',25) ||
				  'Degree of parallelism: requested '  || r_row_stats.pq_requested_degree ||
				  ', actual ' || r_row_stats.pq_actual_degree || ': ' ||
				  r_row_stats.start_timestr || ' - ' || TO_CHAR(SYSDATE,'HH24:MI:SS') );
			EXCEPTION
				WHEN OTHERS THEN
					log_end(r_row_stats.thread_id, SQLERRM);
					PIPE ROW('sid=' || r_row_stats.sid || ' serial#=' || r_row_stats.serial# || ': ' || SQLERRM);
			END;
		END LOOP;

		IF r_row_stats.rowcount = 0 THEN
			RAISE_APPLICATION_ERROR
			( -20000
			, 'Cursor returned no rows' );
		END IF;

		RETURN;
	END pq_submit;


	PROCEDURE submit
		( p_job1 VARCHAR2
		, p_job2 VARCHAR2
		, p_job3 VARCHAR2
		, p_job4 VARCHAR2 )
	IS
		v_results VARCHAR2_TT;
	BEGIN
		SELECT /*+ PARALLEL(4) */ column_value BULK COLLECT INTO v_results
		FROM TABLE(
		     parallel_launch.pq_submit
		     ( VARCHAR2_TT(p_job1,p_job2,p_job3,p_job4)
		     , CURSOR(SELECT thread_id FROM pq_driver)
		     )
		);

		IF v_results.COUNT = 0 THEN
			v_results.EXTEND;
			v_results(1) := 'parallel_launch.SUBMIT: No rows returned from table function PQ_SUBMIT';
		END IF;

		FOR i IN v_results.FIRST..v_results.LAST LOOP
			DBMS_OUTPUT.PUT_LINE(v_results(i));
		END LOOP;
	END submit;


	PROCEDURE test
	IS
	BEGIN
		submit
		( 'parallel_launch.slow_proc(1)'
		, 'parallel_launch.slow_proc(2)'
		, 'parallel_launch.slow_proc(3)'
		, 'parallel_launch.slow_proc(4)' );
	END test;


	PROCEDURE slow_proc
		( p_id INTEGER )
	IS
	BEGIN
		dbms_lock.sleep(2);
	END slow_proc;
END parallel_launch;
/

show errors

prompt Executing parallel_launch.test():
prompt Submits four PL/SQL procedure calls using parallel_launch.submit()
prompt For the test, this uses parallel_launch.slow_proc() which is just a call to DBMS_LOCK.SLEEP(2).
prompt Four 2-second sleep() procedures run end to end should take 8 seconds:
set timi on
exec parallel_launch.test
set timi off

-- Got these results in 10.1.0.3.0:
-- sid=439 serial#=58467:   Degree of parallelism: requested 4, actual 4: 16:18:57 - 16:18:59
-- sid=439 serial#=58467:   Degree of parallelism: requested 4, actual 4: 16:18:59 - 16:19:01
-- sid=439 serial#=58467:   Degree of parallelism: requested 4, actual 4: 16:19:01 - 16:19:03
-- sid=439 serial#=58467:   Degree of parallelism: requested 4, actual 4: 16:19:03 - 16:19:05
--
-- Looks as though I got my PQ slaves but used them one at a time anyway (note start/end times). I don't get it.
-- db_file_multiblock_read_count was 16: maybe CBO decided it can read all of PQ_DRIVER table in one go?
-- (Would it do that for a partitioned table?)
-- parallel_query_mode enabled (default)
-- parallel_threads_per_cpu: 2 (default) and more than 1 CPU on DB server

-- Update 2008-08-01:
-- Possibly if I could repeat the 10.1 test (which I can't now) I would find a PX COORDINATOR FORCED SERIAL 
-- somewhere in the trace; i.e. I forced degree 4 so it generated a parallel plan but could only execute it
-- serially, perhaps feeling that was better than not executing it at all.

-- In home single-CPU 11.1.0.6.0 EE system with parallel_threads_per_cpu = 4 (the default is 2) and
-- taking care to lock the statistics on table pq_driver, I get these results:
--
-- sid=129 serial#=446:     Degree of parallelism: requested 4, actual 4: 19:19:25 - 19:19:27
-- sid=128 serial#=1097:    Degree of parallelism: requested 4, actual 4: 19:19:25 - 19:19:27
-- sid=122 serial#=119:     Degree of parallelism: requested 4, actual 4: 19:19:25 - 19:19:27
-- sid=126 serial#=289:     Degree of parallelism: requested 4, actual 4: 19:19:25 - 19:19:27
--
-- PL/SQL procedure successfully completed.
--
-- Elapsed: 00:00:02.14