|
|
Reading data across sessions using DBMS_PIPE package
Posted On November 8, 2006 by Raja Kishore Reddy filed under
The DBMS_PIPE package allows two or more sessions in the same instance to communicate with each other. Information sent through Oracle pipes is buffered in SGA. All information in pipes is lost when the instance is shut down. Pipes are of two types:
* Public Pipes; and
* 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 disappears when it no longer contains data.
A public explicit pipe is created by calling the function CREATE_PIPE, 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 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; and
* 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, which adds the message to the session's local message buffer. Information in this buffer is sent by calling the SEND_MESSAGE function and 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 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
However, once buffered information is read by a user utilizing 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, including:
* 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 with 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 a single DBMS transaction.
Procedures and Function defined in DBMS_PIPE Package
If we describe the DBMS_PIPE package, the following information is obtained:
SQL> desc dbms_pipe
FUNCTION CREATE_PIPE RETURNS NUMBER(38)
Argument Name Type In/Out Default?
------------------------------ ----------------------- ------ --------
PIPENAME VARCHAR2 IN
MAXPIPESIZE NUMBER(38) IN DEFAULT
PRIVATE BOOLEAN IN DEFAULT
FUNCTION NEXT_ITEM_TYPE RETURNS NUMBER(38)
PROCEDURE PACK_MESSAGE
Argument Name Type In/Out Default?
------------------------------ ----------------------- ------ --------
ITEM VARCHAR2 IN
PROCEDURE PACK_MESSAGE
Argument Name Type In/Out Default?
------------------------------ ----------------------- ------ --------
ITEM NUMBER IN
PROCEDURE PACK_MESSAGE
Argument Name Type In/Out Default?
------------------------------ ----------------------- ------ --------
ITEM DATE IN
PROCEDURE PACK_MESSAGE_RAW
Argument Name Type In/Out Default?
------------------------------ ----------------------- ------ --------
ITEM RAW IN
PROCEDURE PACK_MESSAGE_ROWID
Argument Name Type In/Out Default?
------------------------------ ----------------------- ------ --------
ITEM ROWID IN
ROCEDURE PURGE
Argument Name Type In/Out Default?
------------------------------ ----------------------- ------ --------
PIPENAME VARCHAR2 IN
FUNCTION RECEIVE_MESSAGE RETURNS NUMBER(38)
Argument Name Type In/Out Default?
------------------------------ ----------------------- ------ --------
PIPENAME VARCHAR2 IN
TIMEOUT NUMBER(38) IN DEFAULT
FUNCTION REMOVE_PIPE RETURNS NUMBER(38)
Argument Name Type In/Out Default?
------------------------------ ----------------------- ------ --------
PIPENAME VARCHAR2 IN
PROCEDURE RESET_BUFFER
FUNCTION SEND_MESSAGE RETURNS NUMBER(38)
Argument Name Type In/Out Default?
------------------------------ ----------------------- ------ --------
PIPENAME VARCHAR2 IN
TIMEOUT NUMBER(38) IN DEFAULT
MAXPIPESIZE NUMBER(38) IN DEFAULT
FUNCTION UNIQUE_SESSION_NAME RETURNS VARCHAR2
PROCEDURE UNPACK_MESSAGE
Argument Name Type In/Out Default?
------------------------------ ----------------------- ------ --------
ITEM VARCHAR2 OUT
PROCEDURE UNPACK_MESSAGE
Argument Name Type In/Out Default?
------------------------------ ----------------------- ------ --------
ITEM NUMBER OUT
PROCEDURE UNPACK_MESSAGE
Argument Name Type In/Out Default?
------------------------------ ----------------------- ------ --------
ITEM DATE OUT
PROCEDURE UNPACK_MESSAGE_RAW
Argument Name Type In/Out Default?
------------------------------ ----------------------- ------ --------
ITEM RAW OUT
PROCEDURE UNPACK_MESSAGE_ROWID
Argument Name Type In/Out Default?
------------------------------ ----------------------- ------ --------
ITEM ROWID OUT
Although each of the above mentioned items have their specific usage, for the time being we will concentrate only on 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 that we are creating. We must use this name when we call 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 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, we have to first 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 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 created 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;
Usage of the parameters pipename and maxpipesize is the same as mentioned in the CREATE_PIPE function. The only new parameter is timeout, which is the time (in seconds) to wait while attempting to place a message on a pipe. 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 does not have the required privilege to do so.
RECEIVE_MESSAGE function
This function copies a message into the local message buffer. To receive the message from a pipe, we first 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 the 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, a small example has been provided to illustrate how to use the DBMS_PIPE package. The sample schema named SCOTT, which I am using, is provided by Oracle at the time of database creation. The example I am giving has been tested on Oracle 8.1.6.0.0.
In this example, I want to fix the rate of increment in the salaries of employees working in an organization. The same rate should be used throughout all the sessions, which have connected to Oracle for updating the employees’ salaries.
Connect Oracle with scott/tiger and run code 1.
Code 1
-- 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 that 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 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 procedures are in the 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 session B. For session B, it will be 0 as its default assigned value. DBMS_PIPE can be used to solve this issue, allowing us to access the data throughout different sessions also.
Here is an actual implementation of the package. Run code 2 and check out the results.
Code 2
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% else it will only be 5%. The increment percentage is assigned to 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 note that at the time of sending the message using SEND_MESSAGE procedure as well as at the time of receiving the message using RECEIVE_MESSAGE, the waiting time is set to 60, which means it will wait to send as well as to receive the message for 60 seconds.
How to Use the package?
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’s response will be like as follows:
pipe creation was : 0
message packed with increment percentage of 10
sending message was 0
Here, 0 indicates that all the executions were successful. If any exception occurs, you will get the related exception number.
Connect another Oracle session through 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
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
The author works as a programmer with CS Software Enterprise Ltd., Hyderabad. He can be reached at: ayanmitra_10@rediffmail.com or ayan_mitra@hotmail.com.