Reading data across sessions by using DBMS_PIPE package

As we know Oracle uses PGA (Program Global Area) for execution of PL/SQL programs and SGA (System Global Area) for maintaining other data for a specific session. Several times it is required to share the value of a variable across different sessions. Normally when we use a variable, Oracle de-allocates it as soon as the session ends and at the same time the value of the variable is accessible through out the session only. But what if we want to use the value of the same variable across several sessions for the same instance? The answer is the DBMS_PIPE package.

The DBMS_PIPE package lets two or more sessions in the same instance to communicate with each other. Information sent through Oracle pipes is buffered in the system global area (SGA). All information in pipes is lost when the instance is shut down. 

Pipes are of two types:
· Public Pipes
· Private Pipes 

Public Pipes 

We can create a public pipe either implicitly or explicitly. An implicit public pipe is automatically created when it is referenced for the first time, and it disappears when it no longer contains data. 

A public explicit pipe can be created by calling the CREATE_PIPE function with the private flag set to FALSE. We have to de-allocate explicitly-created pipes by calling the REMOVE_PIPE function. 

Private Pipes

We can explicitly create a private pipe by calling the CREATE_PIPE function. Once created, the private pipe persists in shared memory until we explicitly de-allocate it by calling the REMOVE_PIPE function. A private pipe is also de-allocated when the database instance is shut down. However we cannot create a private pipe if an implicit pipe exists in memory and has the same name as the private pipe we are trying to create. In this case, CREATE_PIPE returns an error. 

Access to a private pipe is restricted to: 
· Sessions running under the same User ID as the creator of the pipe 
· Stored subprograms executing in the same User ID privilege domain as the pipe creator 
· Users connected as SYSDBA or INTERNAL 

The Approach

The sending session builds a message by using one or more calls to the PACK_MESSAGE procedure. This procedure adds the message to the session's local message buffer. The information in this buffer is sent by calling the SEND_MESSAGE function, designating the pipe name to be used to send the message. When SEND_MESSAGE is called, all messages that have been stacked in the local buffer are sent. 

A process that wants to receive a message calls the RECEIVE_MESSAGE function, designating the pipe name from which to receive the message. The process then calls the UNPACK_MESSAGE procedure to access each of the items in the message. 

Each public pipe works asynchronously. Any number of schema users can write to a public pipe, as long as they have EXECUTE permission on the DBMS_PIPE package, and they know the name of the public pipe. To grant this privilege to your schema user simple execute the following SQL grant statement after connecting as SYS:

GRANT EXECUTE ON DBMS_PIPE TO <user_name>

However, once buffered information is read by one user by using RECEIVE_MESSAGE and then unpacked by UNPACK_MESSAGE, it is emptied from the buffer, and is not available for other readers of the same pipe. 

Pipe Uses

The pipe functionality has several potential applications: 
· External service interface: We can communicate with user-written services that are external to the RDBMS. This can be done in a (effectively) multi-threaded manner, so that several instances of the service are executing simultaneously. Additionally, the services are available asynchronously. The requestor of the service does not need to block a waiting reply. The requestor can check (with or without timeout) at a later time. 
· Independent transactions: The pipe can communicate to a separate session which can perform an operation in an independent transaction.
· Debugging: Triggers and stored procedures can send debugging information to a pipe. Another session can keep reading out of the pipe and display it on the screen or write it to a file. 
· Concentrator: This is useful for multiplexing large numbers of users over a fewer number of network connections, or improving performance by concentrating several user-transactions into one DBMS transaction. 

Procedures and Function defined in DBMS_PIPE Package
If we describe the DBMS_PIPE package we can get the following information:

SQL> desc dbms_pipe

Although each of the above mentioned items is having their specific usage, but for the time being we will concentrate only some of them:

CREATE_PIPE function

This function explicitly creates a public or private pipe. If the private flag is TRUE, then the pipe creator is assigned as the owner of the private pipe. 

Syntax:
DBMS_PIPE.CREATE_PIPE (
pipename IN VARCHAR2,
maxpipesize IN INTEGER DEFAULT 8192,
private IN BOOLEAN DEFAULT TRUE)
RETURN INTEGER;

Here pipename is the name of the pipe which we are creating. We must use this name when we are calling SEND_MESSAGE and RECEIVE_MESSAGE. This name must be unique across the instance. 
The maxpipesize is the size for the pipe, in bytes.The total size of all of the messages on the pipe cannot exceed this amount. The message is blocked if it exceeds this maximum. The default maxpipesize is 8192 bytes. 
The private uses the default, TRUE, to create a private pipe

PACK_MESSAGE procedure
This procedure builds our message in the local message buffer. To send a message, first we have to make one or more calls to PACK_MESSAGE. Then, we have to call SEND_MESSAGE to send the message in the local buffer on the named pipe. 
The PACK_MESSAGE procedure is overloaded to accept items of type VARCHAR2, NUMBER, or DATE. When we call SEND_MESSAGE to send this message, we must indicate the name of the pipe on which we want to send the message. If this pipe already exists, then we must have sufficient privileges to access this pipe. If the pipe does not already exist, then it is created automatically. 

Syntax:
DBMS_PIPE.PACK_MESSAGE (item IN VARCHAR2);
DBMS_PIPE.PACK_MESSAGE (item IN NCHAR);
DBMS_PIPE.PACK_MESSAGE (item IN NUMBER);
DBMS_PIPE.PACK_MESSAGE (item IN DATE);
DBMS_PIPE.PACK_MESSAGE_RAW (item IN RAW);
DBMS_PIPE.PACK_MESSAGE_ROWID (item IN ROWID);

SEND_MESSAGE function

This function sends a message on the named pipe. The message is contained in the local message buffer, which was filled with calls to PACK_MESSAGE. A pipe could be explicitly using CREATE_PIPE; otherwise, it is created implicitly. 

Syntax:
DBMS_PIPE.SEND_MESSAGE (
pipename IN VARCHAR2,
timeout IN INTEGER DEFAULT MAXWAIT,
maxpipesize IN INTEGER DEFAULT 8192)
RETURN INTEGER;

The usage of the parameters pipename and maxpipesize are same as mentioned in the CREATE_PIPE function. The only new parameter is timeout, which is the time to wait while attempting to place a message on a pipe, in seconds. The default value is the constant MAXWAIT, which is defined as 86400000 (1000 days). 

The SEND_MESAGE returns 0 for success, 1 if Timed out, 3 if an interrupt occurred and ORA-23322 if the user who is trying to send the message not having sufficient privilege to do so.

RECEIVE_MESSAGE function

This function copies the message into the local message buffer. To receive a message from a pipe, first we have to call RECEIVE_MESSAGE. When we receive a message, it is removed from the pipe; hence, a message can only be received once. 

Syntax:
DBMS_PIPE.RECEIVE_MESSAGE (
pipename IN VARCHAR2,
timeout IN INTEGER DEFAULT maxwait)
RETURN INTEGER;

All the parameters and values returned are same as discussed previously. 

UNPACK_MESSAGE procedure

This procedure retrieves items from the buffer. 

Syntax:
DBMS_PIPE.UNPACK_MESSAGE (item OUT VARCHAR2);
DBMS_PIPE.UNPACK_MESSAGE (item OUT NCHAR);
DBMS_PIPE.UNPACK_MESSAGE (item OUT NUMBER);
DBMS_PIPE.UNPACK_MESSAGE (item OUT DATE);
DBMS_PIPE.UNPACK_MESSAGE_RAW (item OUT RAW);
DBMS_PIPE.UNPACK_MESSAGE_ROWID (item OUT ROWID);


Here I am giving a small example on how to use the DBMS_PIPE package. The sample schema named SCOTT I am using is provided by Oracle at the time of database creation. The example I am giving is tested on Oracle 8.1.6.0.0.

In the example I want to fix a rate of increment in their salary for the employees working in an organization. The same rate should be used throughout all the sessions which have connected Oracle for updating the salary of the employees.

Connect Oracle with scott/tiger and run the following code snippet.

-- Creation of emp_inc_rate package
create or replace package emp_inc_rate
as 
sal_inc_per number(5,2) :=0;
procedure set_inc_rate(yr in number);
procedure get_per;
end;
/

Please do note here a package description has been created which has defined two procedures and a variable. The first procedure called set_inc_rate will fix the rate of the increment in the salary depending upon the year passed as an argument. The rate fixed will be available in the sal_inc_per variable which the second procedure get_per will read and update the salaries accordingly. Although both the procedures are in same package we can directly use the sal_inc_per variable, but strictly if it is running in the same session. For example if session A has assigned the sal_inc_per to 10% then normally it will not be available for the session B. For session B it will be 0 as its default assigned value. Using DBMS_PIPE can solve the purpose which will allow us to access the data throughout different sessions also.

Here I am giving the actual implementation of the package. Run the following code snippet.

create or replace package body emp_inc_rate
as
procedure set_inc_rate(yr in number) as
cur_yr number(4):=yr; 
pm integer;
sm integer;
begin
if mod(cur_yr,5)=0
then
sal_inc_per:=10;
else
sal_inc_per:=5;
end if;

pm := dbms_pipe.create_pipe('sal_inc',8192,true);
dbms_output.put_line('pipe creation was : '|| pm);

dbms_pipe.pack_message(sal_inc_per);
dbms_output.put_line('message packed with increment percentage of ' || sal_inc_per);

sm :=dbms_pipe.send_message('sal_inc',60,8192);
dbms_output.put_line('sending message was ' || sm);

end set_inc_rate;

procedure get_per 
as
cur_per number(5,2);
pm integer;
begin
pm :=dbms_pipe.receive_message('sal_inc',60);
dbms_output.put_line('receiving message was ' || pm);

dbms_pipe.unpack_message(cur_per);
dbms_output.put_line('current increment rate is fixed to ' || cur_per);

dbms_output.put_line('incrementing the salaries');
update emp set sal=sal+(sal*(cur_per/100));
dbms_output.put_line('salaries updated');
end get_per;
end;
/

The first procedure set_inc_rate has fixed the rate of interest depending upon the year passed to it. For example if the year is passed is 5th year of any decade then the increment will be 10% otherwise it will be only 5%. The increment percentage is assigned to the sal_inc_per variable. After that it has created a pipe called sal_inc and packed the same increment percentage and sent it. 
The second procedure get_per received the incremented rate and unpacked the information. It stored the message in a variable call cur_per and updated the salaries accordingly.

Please also note at the time of sending the message using the SEND_MESSAGE procedure as well as at the time of receiving the message using the RECEIVE_MESSAGE, the waiting time is set to 60, which means it will wait the send as well as to receive the message for 60 seconds (1 minute)

How to Use
Connect an Oracle sessions by scott user account. Run the following commands in one of the sessions (let us call it session A): 

SQL> set serveroutput on
SQL> set feedback off
SQL> execute emp_inc_rate.set_inc_rate(2005)

Oracle responds like this:

pipe creation was : 0
message packed with increment percentage of 10
sending message was 0

Here 0 indicates all the executions were successful. If any exception occurs you will be getting the related exception number.

Connect another Oracle session by the same user account (scott) and run the following commands:

SQL> set serveroutput on
SQL> set feedback off
SQL> Select ename, sal from emp where rownum<3;

Oracle responds like this:

ENAME SAL 
------------------------------ -------------
ADAMS 5000.00
BLAKE 10000.00

The output may differ depending upon the rows in the emp table. Now execute the following command:

SQL> execute emp_inc_rate.get_per

Oracle responds like this:

receiving message was : 0
current increment rate is fixed to 10
incrementing the salaries
salaries updated

Now check the emp table by passing the following command:

SQL> Select ename, sal from emp where rownum<3;

Oracle responds like this:

ENAME SAL 
------------------------------ -------------
ADAMS 5500.00
BLAKE 11000.00